CrewAI

Install the integration extra:

pip install "agentx-python[crewai]" crewai

`observer.kickoff()`

Wraps crew.kickoff() for you. Task outputs are automatically captured as tool calls.

from agentx import AgentX
from agentx.integrations.crewai import AgentXCrewObserver
from crewai import Agent, Task, Crew

client = AgentX.from_env()  # AGENTX_API_KEY (+ AGENTX_API_BASE_URL for self-host)
observer = AgentXCrewObserver(
    tracer=client.tracer,
    name="support-crew",
    metadata={"env": "production"},
)

researcher = Agent(
    role="Policy Researcher",
    goal="Find relevant company policies",
    backstory="You look up policies from the knowledge base.",
    verbose=False,
)
writer = Agent(
    role="Support Writer",
    goal="Write clear, friendly support responses",
    backstory="You turn policy details into helpful customer replies.",
    verbose=False,
)
crew = Crew(
    agents=[researcher, writer],
    tasks=[
        Task(
            description="Look up the cancellation policy and summarize it.",
            expected_output="A concise policy statement.",
            agent=researcher,
        ),
        Task(
            description="Write a friendly support reply about cancelling a subscription.",
            expected_output="A clear, helpful response to send to the customer.",
            agent=writer,
        ),
    ],
    verbose=False,
)

result = observer.kickoff(crew, inputs={"query": "cancel subscription"})
print(result.raw)

client.tracer.flush(timeout=10)

What gets traced

Field	Source
`input`	`inputs` dict passed to `kickoff()`
`output`	`result.raw` (CrewOutput string)
`latencyMs`	Wall-clock time of the full crew run
`toolCalls`	Each task output, with `description` as `name` and `raw` as `output`
`performanceSummary`	One execution step per task, with real start/end timestamps from CrewAI’s event bus
`error`	Exception message if `crew.kickoff()` raises, or a task-level error from `TaskFailedEvent`

Per-task timing comes from CrewAI’s event bus (TaskStartedEvent / TaskCompletedEvent / TaskFailedEvent, available on modern CrewAI versions) - observer.kickoff() registers temporary listeners for the duration of the call, keyed by each task’s real id so concurrent tasks (async_execution=True) are timed correctly rather than attributed to whichever task started most recently.

On older CrewAI versions that predate the event bus, kickoff() falls back to splitting the total latency evenly across tasks instead - still useful for task order, just not exact per-task duration on those versions.

`observer.observe()` context manager

Use this instead of kickoff() when you need to call crew.kickoff() yourself, for example to catch exceptions before they reach AgentX, or to run additional logic between kickoff and reading the result:

from agentx import AgentX
from agentx.integrations.crewai import AgentXCrewObserver
from crewai import Agent, Task, Crew

client = AgentX.from_env()
observer = AgentXCrewObserver(tracer=client.tracer, name="support-crew")

researcher = Agent(
    role="Policy Researcher",
    goal="Find relevant company policies",
    backstory="You look up policies from the knowledge base.",
)
crew = Crew(
    agents=[researcher],
    tasks=[
        Task(
            description="Look up the cancellation policy and summarize it.",
            expected_output="A concise policy statement.",
            agent=researcher,
        )
    ],
)

with observer.observe(name="support-crew", input={"query": "cancel subscription"}) as span:
    try:
        result = crew.kickoff(inputs={"query": "cancel subscription"})
        span.output = result.raw
    except Exception as e:
        span.set_error(str(e))
        raise

client.tracer.flush(timeout=10)

Unlike kickoff(), the observe() context manager doesn’t auto-populate toolCalls or performanceSummary from task outputs. Record them yourself with span.add_tool_call(...) if you need that detail.

`AgentXCrewObserver` reference

AgentXCrewObserver(
    tracer: Tracer,
    name: str = "crewai-crew",
    metadata: dict | None = None,
    session_id: str | None = None,
)

Parameter	Description
`tracer`	`client.tracer` from your `AgentX` instance
`name`	Label shown in the AgentX UI
`metadata`	Static key-value metadata attached to every trace
`session_id`	Links traces from the same conversation thread

`observe()` parameters

observer.observe(
    name: str | None = None,
    input: Any = None,
    metadata: dict | None = None,
    session_id: str | None = None,
) -> contextmanager

Inside the with block, assign span.output before exiting to set the trace output.

Call tracer.flush() before your process exits in scripts or one-shot jobs. In long-running servers it is not required: traces drain automatically in the background.

Tracing

Framework Integrations

`observer.kickoff()`

What gets traced

`observer.observe()` context manager

`AgentXCrewObserver` reference

`observe()` parameters

​observer.kickoff()

​What gets traced

​observer.observe() context manager

​AgentXCrewObserver reference

​observe() parameters

`observer.kickoff()`

What gets traced

`observer.observe()` context manager

`AgentXCrewObserver` reference

`observe()` parameters