Instrument an OpenClaw agent (5-minute quickstart)

1. Get a connector + bearer token

1. Open the Dobby dashboard and pick a tenant.
2. Click Workloads → Connect → Dobby SDK, choose Python, and pick OpenClaw as the framework.
3. Copy the connector_id (wc_...) and api_key (dsdk_...) — the api_key is shown ONCE, store it in your secrets manager immediately.

2. Install

pip install "dobby-collector[openclaw]"

The [openclaw] extra pulls openclaw-sdk >= 2.1.0 as a transitive dep. If you already have it in your venv, the base dobby-collector package is enough.

3. Instrument your agent

import asyncio
from openclaw_sdk import OpenClawClient
from dobby_collector import init
from dobby_collector.integrations.openclaw import instrument_openclaw

# 1. Initialise the Dobby SDK once at process startup
init(
    api_key="dsdk_...",          # from your Dobby Workloads page
    connector_id="wc_...",        # ditto
    base_url="https://dobby-ai.com",
)

# 2. Instrument openclaw-sdk ONCE — patches Agent.execute /
#    Agent.execute_stream_typed. OpenClaw does NOT surface the model on a
#    run, so thread it in via default_model (same value CostTracker needs).
instrument_openclaw(default_model="claude-sonnet-4-6")

# 3. Drive your OpenClaw agent as usual — every run is captured automatically
async def main():
    async with OpenClawClient.connect() as client:
        agent = client.get_agent("incident-triage")
        result = await agent.execute("Summarise today's incidents")
        print(result.content)
        # → one workload_run in Dobby: model + token usage + tool calls

asyncio.run(main())

• One init + one instrument call per process. Both are idempotent — calling twice is a no-op. uninstrument_openclaw() restores the originals.
• One run per execute(). Concurrent runs (e.g. asyncio.gather) are isolated — no shared run state.
• Non-blocking. Events buffer in-process; a background thread ships them every 10s (or sooner). Instrumentation NEVER raises into your agent — failures are logged and swallowed.

Streaming is supported too:

# Streaming works too — the typed event stream is captured as it arrives,
# and every event still passes through to your code unchanged.
# Note: execute_stream_typed is an async generator — iterate it directly, no await.
async for ev in agent.execute_stream_typed("Draft the incident report"):
    ...  # your handling — Dobby observes content / tool_call / tool_result / done

4. Production: use env vars

# Production: read from env vars instead of hardcoding
export DOBBY_API_KEY="dsdk_..."
export DOBBY_CONNECTOR_ID="wc_..."
export DOBBY_BASE_URL="https://dobby-ai.com"

# Then in code:
from dobby_collector import init
init()   # auto-reads DOBBY_* env vars

5. Verify it worked

Within ~10 seconds of agent.execute() completing, a new run appears on Workloads → Runs. Click into it to see the per-event timeline (tool calls, model, token usage). For programmatic verification in BigQuery:

SELECT
  external_run_id,
  status,
  JSON_VALUE(metadata_json, '$.framework')    AS framework,      -- "openclaw"
  JSON_VALUE(metadata_json, '$.sdk_version')  AS sdk_version,    -- "0.6.0" or newer
  JSON_VALUE(metadata_json, '$.llm_calls[0].model') AS model,    -- your default_model
  ARRAY_LENGTH(JSON_QUERY_ARRAY(metadata_json, '$.tool_calls')) AS tool_calls,
  ARRAY_LENGTH(JSON_QUERY_ARRAY(metadata_json, '$.llm_calls'))  AS llm_calls
FROM `workload_run_payloads`
WHERE connector_id = 'wc_...'
ORDER BY created_at DESC
LIMIT 5

Expected on a successful run: framework="openclaw", model = your default_model, llm_calls = 1, and tool_calls matching the tools your agent used.

What the integration captures

Troubleshooting

Related

• CrewAI (OSS) quickstart — the same SDK, CrewAI handler
• Python Gateway SDK — synchronous LLM/MCP policy enforcement
• OpenClaw — upstream project