How do I trace an agent built with CrewAI and deployed to a Celery worker?

Rius traces a CrewAI crew when you install the OpenAI integration, call rius.init in each worker process after it forks, and wrap your entrypoint and tool bodies with observe. Name the agent with service_name and the name on your root observe wrapper. Celery’s worker processes skip Python’s exit hook, so you flush and shut the client down from the worker_process_shutdown signal.

How it works

• Model calls. CrewAI 1.x drives the OpenAI client natively, so the openai integration captures its model calls as generations with model, tokens, cost and time to first token. If LiteLLM is installed, CrewAI uses it instead, and you use the litellm integration. See CrewAI and the Integrations matrix.
• Run structure. Crews and tasks produce no spans of their own, so your observe wrappers give the run its shape. A root wrapper of kind AGENT around crew.kickoff() becomes the root span. A wrapper of kind TOOL on each tool body becomes a tool span under it.
• Export. Spans are queued in memory and sent in batches from a background thread, so tracing never sits on your agent’s hot path.
• Forked workers. Celery’s prefork pool forks its worker processes and ends them through os._exit, which skips Python’s exit hook. So start Rius in each child from worker_process_init and shut it down from worker_process_shutdown. Each child gets its own heartbeat instance ID, so the Agents page shows one instance per worker process.

Set it up

1. Install the SDK with the OpenAI extra, next to CrewAI. The extra installs the instrumentation, not the OpenAI library, which CrewAI already brings. Installation is covered in full on the CrewAI page.

   pip install "glassflow-rius[openai]" crewai

2. Call rius.init in each worker process, never at module scope, because the parent process imports your tasks module too. Decide the agent’s names there. service_name is the name the agent is grouped under, because RIUS_AGENT_NAME defaults to the service name. Name the root wrapper too, since the root span takes the name you pass to observe. Decorator order on tools matters: @tool outermost, observe on the function body.

   import rius
   from celery import Celery
   from celery.signals import worker_process_init, worker_process_shutdown
   from crewai import Agent, Crew, Task
   from crewai.tools import tool
   from rius import SpanKind
    
   app = Celery("agents", broker="redis://localhost:6379/0")
   client = None
    
   @worker_process_init.connect
   def start_rius(**kwargs):
       global client
       client = rius.init(api_key="ri_...", service_name="my-agent")
    
   @worker_process_shutdown.connect
   def stop_rius(**kwargs):
       if client is not None:
           client.flush(timeout_millis=10_000)  # force an export
           client.shutdown()  # drain, stop the thread, send the stopped heartbeat
    
   @tool("account_status")
   @rius.observe(kind=SpanKind.TOOL)
   def account_status(email: str) -> str:
       """Fetch a customer's account standing by email."""
       return lookup(email)
    
   support_agent = Agent(
       role="Support specialist",
       goal="Answer support tickets accurately and concisely.",
       backstory="You resolve product and account questions.",
       tools=[account_status],
       llm="gpt-5-mini",
   )
    
   @app.task
   @rius.observe(name="handle-ticket", kind=SpanKind.AGENT)
   def handle_ticket(ticket: str) -> str:
       task = Task(
           description=f"Resolve this support ticket: {ticket}",
           expected_output="A short, direct answer.",
           agent=support_agent,
       )
       return Crew(agents=[support_agent], tasks=[task]).kickoff().raw

   The threads, gevent and eventlet pools do not send worker_process_init, so with those pools call init once when the worker starts.

3. Keep CrewAI’s own tracing prompt out of a process with no terminal by setting CREWAI_TRACING_ENABLED=false in the worker’s environment. CrewAI otherwise asks about its tracing product interactively on first run.

4. Group each job’s traces by reusing your own job or conversation id with rius.session() around the kickoff. See Integrations for the rest of the setup options.

5. Flush only at the process boundary, as the shutdown handler does. Do not call flush() inside a task, because it blocks until the export completes. A worker process killed with SIGKILL, by a hard time limit or the out-of-memory killer, loses the spans still queued in the batch and sends no stopped heartbeat. Reliability covers batching, retries and shutdown.

6. Run one ticket that calls a tool, then open the trace in the console.

What you get

• A root span of kind AGENT named handle-ticket, or whatever name you gave the wrapper.
• An LLM generation for each model call, with model, tokens, cost and time to first token. Cost is a list-price estimate in US dollars that Rius computes, so you never send it.
• A TOOL span for each tool, named after the tool, parented under the run.
• Spans from one job grouped in a session, and an agent on the Agents page under your service_name, with one instance per process while the heartbeat is on.
• Generation spans carry the generic name ChatCompletion. Crews and tasks have no spans, so which task a generation belongs to is not visible unless you wrap per-task entrypoints with observe.

When something else fits

• If LiteLLM is installed in the worker’s environment, use the litellm integration instead of openai.
• If you need to see which task a model call belongs to, add an observe wrapper around each task’s entrypoint.
• If workers run with the heartbeat off or without credentials, the agent is marked no heartbeat and lists no instances, though traces still arrive.

Related