Instrument a CrewAI Crew (5-minute quickstart)

When to use the SDK vs other modes

The SDK and the LLM/MCP Gateways aren't mutually exclusive — combine for defense-in-depth. SDK gives you span-level workload runs; gateways give you synchronous policy enforcement.

1. Get a connector + bearer token

1. Open the Dobby dashboard and pick a tenant.
2. Click Workloads → Connect → Dobby SDK. Pick "Python (paste credentials)".
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[crewai]"

The [crewai] extra pulls crewai >= 1.0 as a transitive dep. If you already have CrewAI installed in your venv, the base dobby-collector package is enough. Verified against crewai==1.14.4.

3. Instrument your crew

from crewai import Agent, Crew, Process, Task
from crewai.tools import tool
from dobby_collector import init
from dobby_collector.integrations.crewai import DobbyCrewAIHandler

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

# 2. Register the handler ONCE — it auto-listens on the global event bus
handler = DobbyCrewAIHandler()    # auto_run=True by default

# 3. Your existing CrewAI code runs unchanged
@tool("get_current_date")
def get_current_date() -> str:
    """Returns the current ISO-8601 UTC date."""
    from datetime import datetime, timezone
    return datetime.now(timezone.utc).date().isoformat()

researcher = Agent(
    role="Researcher",
    goal="Fetch today's date using the tool",
    backstory="Precise, tool-only.",
    tools=[get_current_date],
)
writer = Agent(role="Writer", goal="Summarise the date in 1 sentence", backstory="Concise.")

research_task = Task(
    description="Call get_current_date and return the ISO string.",
    expected_output="An ISO-8601 date string.",
    agent=researcher,
)
summarise_task = Task(
    description="Write 'Today is <weekday> <month> <day>, <year>.'",
    expected_output="A single sentence.",
    agent=writer,
    context=[research_task],
)

crew = Crew(
    agents=[researcher, writer],
    tasks=[research_task, summarise_task],
    process=Process.sequential,
)
result = crew.kickoff()   # every span, LLM call, tool call → Dobby automatically

• One init call per process. Idempotent — safe to call multiple times, returns the existing instance.
• One handler instance globally. Registers itself on the global crewai_event_bus singleton in __init__ — no manual wire-up.
• Auto run boundaries. auto_run=True (default) wraps each crew.kickoff() in start_run/end_run. Set DobbyCrewAIHandler(auto_run=False) if you manage runs yourself.
• Thread-safe + non-blocking. Events buffer in-process; a background thread ships them every 10s (or sooner if buffer fills). The handler NEVER raises into your crew code — failures are logged and swallowed.

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 crew.kickoff() completing, a new run appears on Workloads → Runs. Click into it to see the per-event timeline (task spans, agent execution, every LLM + tool call). For programmatic verification in BigQuery:

SELECT
  external_run_id,
  status,
  JSON_VALUE(metadata_json, '$.framework')    AS framework,      -- "crewai"
  JSON_VALUE(metadata_json, '$.sdk_version')  AS sdk_version,    -- "0.4.1" or newer
  JSON_VALUE(metadata_json, '$.traceparent')  AS traceparent,    -- W3C
  ARRAY_LENGTH(JSON_QUERY_ARRAY(metadata_json, '$.tool_calls')) AS tool_calls,
  ARRAY_LENGTH(JSON_QUERY_ARRAY(metadata_json, '$.llm_calls'))  AS llm_calls,
  ARRAY_LENGTH(JSON_QUERY_ARRAY(metadata_json, '$.agent_steps')) AS agent_steps
FROM `workload_runs`
WHERE connector_id = 'wc_...'
ORDER BY started_at DESC
LIMIT 5

Expected on a successful crew: framework="crewai", sdk_version="0.4.0" (or current), traceparent populated, tool_calls > 0 and llm_calls > 0.

What the handler captures

Troubleshooting

Related

• CrewAI Cloud / Enterprise connector — managed platform, webhook delivery, no SDK
• Distributed tracing (W3C traceparent) — how the SDK auto-emits W3C context
• Python Gateway SDK — synchronous LLM/MCP policy enforcement
• CrewAI docs — upstream framework reference