How do I trace an agent built with the OpenAI Agents SDK and deployed to a Celery worker?

Rius traces an OpenAI Agents SDK agent in a Celery worker with the Python SDK, the openai integration and a few observe wrappers. Call rius.init in each worker process from Celery’s worker_process_init signal, shut the client down from worker_process_shutdown, wrap the entrypoint and tool bodies, and name the agent with underscores.

How it works

The OpenAI Agents SDK drives the OpenAI client underneath, so the openai integration captures every model call as a generation with model, tokens, cost and time to first token. With a global init(), every installed integration is enabled automatically, and there is no framework-specific code to add. See Integrations for the full list.

The framework keeps two things to itself, so each needs one line from you:

• Run structure. The agent loop, tool executions and handoffs run inside the framework. Wrapping the entrypoint and the tool bodies with observe puts that structure in the trace.
• Built-in tracing. The framework uploads its own traces to the OpenAI platform, independent of OpenTelemetry. set_tracing_disabled(True) makes Rius the single telemetry path. Leaving it on does not interfere with Rius.

Spans are queued in your process and exported in batches from a background thread, so the network never sits on the agent’s hot path. 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 after the fork and shut it down yourself before the child exits. Each worker process gets its own heartbeat instance ID and shows as its own instance on the Agents page. The Reliability page covers the pipeline in full.

Set it up

1. Install the SDK with the OpenAI extra, plus the framework. The extra installs the instrumentation, never the openai library itself, so openai must already be in your environment (the Agents SDK pulls it in).

pip install "glassflow-rius[openai]" openai-agents

2. Initialize Rius per worker process, disable the framework’s own tracing, and wrap the tool and the entrypoint. This adapts the example from the OpenAI Agents SDK page to Celery. The service_name you pass to init is the name the agent is grouped under, and RIUS_AGENT_NAME overrides it if you set it.

import rius
from agents import Agent, Runner, function_tool, set_tracing_disabled
from celery import Celery
from celery.signals import worker_process_init, worker_process_shutdown
from rius import SpanKind
 
app = Celery("agents", broker="redis://localhost:6379/0")
set_tracing_disabled(True)
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
 
@function_tool
@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(
    name="support_agent",
    instructions="Answer support questions; check accounts with the tool.",
    tools=[account_status],
    model="gpt-5-mini",
)
 
@app.task
@rius.observe(name="handle-ticket", kind=SpanKind.AGENT)
def handle_ticket(ticket: str) -> str:
    result = Runner.run_sync(support_agent, ticket)
    return str(result.final_output)

3. Keep the decorator order on tools: @function_tool outermost, observe on the function body. Use SpanKind.TOOL there, because a bare observe makes a CHAIN span that the Tool kind filter and the Tool Loop alert never see.

4. Name things deliberately:

   • service_name is the agent’s name in Rius, so choose one stable name for the worker’s agent.
   • Give the Agent an underscore name such as support_agent. Hyphens are rewritten in the generated handoff tool name, and the framework warns on every run.
   • Name the root span with observe(name=...), such as handle-ticket. Model calls from the Responses API are all named Response, so the agent name does not appear on those spans.

5. Keep rius.init out of module scope, because the parent process imports your tasks module too. The threads, gevent and eventlet pools do not send worker_process_init, so with those pools call init once when the worker starts. Never call flush() inside a task, because it blocks until the export completes. Batch export also means a worker process killed with SIGKILL, by a hard time limit or the out-of-memory killer, loses the spans still queued and sends no stopped heartbeat, so its instance reports gone.

6. To group turns of one conversation, scope each run with rius.session() and reuse your conversation or ticket id.

What you get

• A root AGENT span for each run, with a generation for every model call showing 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 named TOOL span for each function tool.
• The agent on the Agents page under your service_name, with one instance for each worker process while heartbeats flow.
• Sessions that group the turns of one conversation.

Handoffs produce no span of their own. They show as consecutive generations that belong to different agents. If handoff boundaries matter, add an observe wrapper per agent entry. Streamed runs carry token counts only if usage is enabled on the underlying client.

When something else fits

• If your Celery worker runs a different agent framework, start from the matching page on Integrations, because the setup differs by framework.
• If you only call the OpenAI client without the Agents SDK, the plain OpenAI integration is enough and you need no framework wrappers.

Related