Monitoring agents in production

Once an agent is live you stop asking “does it work?” and start asking “what did it just do, and what did it cost?”. Agentor answers both from one place: the stream of events every run emits. Those events go two ways. They are uploaded to Celesto as a trace you can click through, and they come back on the result object so your own code can read them.

Send runs to Celesto

Set your Celesto API key and turn tracing on for the agent — a run is only recorded when both are true:

export CELESTO_API_KEY="cel_..."

from agentor import Agentor

agent = Agentor(
    name="Support Agent",
    model="gpt-5-mini",
    tools=["get_weather"],
    enable_tracing=True,
)

result = agent.run("What's the weather in Paris?")
# View the trace at https://celesto.ai/observe

Each run becomes one trace: an agent span, a span per model call, and a span per tool call, with timings and token counts throughout. See Tracing for setup options, private endpoints, and what each span holds.

Tests need no special handling - tracing is off unless you ask for it. To keep one run out of an agent that has it on, pass tracing=False on that call. Runs behave identically either way; only the upload is skipped.

Read a run in code

agent.run() and await agent.arun() return a RunResult. It carries the answer and the evidence:

result = agent.run("Weather in London and Paris?")

result.final_output   # 'It is sunny and 22C in both London and Paris.'
result.status         # 'completed'
result.error          # None
result.run_id         # set when the agent has a store, otherwise None
result.usage          # Usage(input_tokens=195, output_tokens=63, total_tokens=258)
result.messages       # the full conversation, ready to feed back in
result.events         # everything that happened, in order

str | BaseModel | None

The agent’s answer. A parsed model instance when output_type is set, None if the run did not finish.

'completed' | 'max_turns' | 'failed'

How the run ended. Always check this before trusting final_output.

str | None

Why the run did not complete. None on success.

Usage

input_tokens, output_tokens, and total_tokens summed across every model call in the run.

list[dict]

The conversation as the model saw it — user, assistant, and tool messages. Pass it back into arun() to continue the conversation.

list[Event]

Every step: run_start, generation, tool_call, tool_result, message, run_end.

Track token usage

result = agent.run("Summarize today's tickets.")

print(f"{result.usage.total_tokens} tokens "
      f"({result.usage.input_tokens} in, {result.usage.output_tokens} out)")

For per-call detail, walk the generations:

for event in result.events:
    if event.type == "generation" and event.usage:
        print(f"turn {event.turn}: {event.usage.total_tokens} tokens")

See which tools ran

result.tool_calls lists the calls the model asked for, with the arguments it chose:

for call in result.tool_calls:
    print(call.name, call.args)
# get_weather {'city': 'London'}
# get_weather {'city': 'Paris'}

For what each tool actually returned, read the results:

for event in result.events:
    if event.type == "tool_result":
        status = "failed" if event.error else "ok"
        print(f"{event.name} [{status}]: {event.result[:80]}")

Tell the three endings apart

A run that did not produce an answer is not automatically an exception. Check status:

completed
max_turns
failed

The agent answered. final_output is set, error is None.

The agent used its whole turn budget without settling on an answer — usually a loop between tools, or instructions it cannot satisfy.

result = agent.run("Weather in London?")
result.status        # 'max_turns'
result.error         # 'Reached max_turns (2) without a final answer.'
result.final_output  # None

Raise the budget with Agentor(..., max_turns=40), or per call with await agent.arun(prompt, max_turns=40). The default is 20.

Something raised — a provider error, a bad API key, a network failure. The exception propagates to your caller, and the run_end event is still recorded with status="failed" so the trace and the saved run show what happened.

result = agent.run(prompt)

if result.status != "completed":
    logger.warning("Run ended as %s: %s", result.status, result.error)

Failing tools do not kill the run

A tool that raises does not end the run. The error text goes back to the model, which can try something else or explain the gap. If the same tool keeps failing, it is withdrawn after two attempts so it cannot burn the remaining turns:

@function_tool
def flaky_lookup(query: str) -> str:
    """Look up a record in the records database.

    Args:
        query: What to look up.
    """
    raise RuntimeError("database unavailable")


agent = Agentor(name="Lookup Agent", model="gpt-4o-mini", tools=[flaky_lookup])
result = agent.run("Look up the record for customer 42.")

result.status        # 'completed'
result.final_output  # "I'm currently unable to access the database ..."

The tool ran twice, then stopped being offered. Both failures are visible in the run:

failures = [e for e in result.events if e.type == "tool_result" and e.error]
print(len(failures))          # 2
print(failures[0].error)      # RuntimeError: database unavailable

The failure budget defaults to two attempts per tool. Adjust it per agent with Agentor(..., max_tool_failures=3) — higher for tools that are legitimately flaky, lower to fail fast.

Keep a copy of every run

Tracing is for looking at runs. A store is for keeping them — and for finishing them if the process dies:

from agentor.engine.store import FileStore

agent = Agentor(name="Support Agent", tools=[...], store=FileStore("runs"))

result = agent.run("Summarize today's tickets.")
result.run_id  # 'cabc9046ac5040298aa362c0d6d2c5c5'

Every event lands in runs/<run_id>.jsonl as it happens, so you can replay a run long after the dashboard has moved on, and agent.resume(run_id) can pick up an interrupted one.

Watch a run as it happens

For live progress rather than a post-mortem, stream it:

async for event in agent.stream_chat("Research quantum computing", serialize=False):
    if event.tool_action:
        print(f"[{event.tool_action.type}] {event.tool_action.name}")
    elif event.message:
        print(event.message)

See the streaming guide for the full event shape.

Next steps

Tracing

Set up Celesto tracing and learn what each span records.

Durable runs

Save runs to disk and resume them after a crash.

Streaming

Show progress to users while the agent works.

Deployment

Ship the agent, with observability already on.

Start Here

Build Agents

Concepts

Tool Use & MCP

Guides

SuperAuth

Monitoring agents in production

Send runs to Celesto

Read a run in code

Track token usage

See which tools ran

Tell the three endings apart

Failing tools do not kill the run

Keep a copy of every run

Watch a run as it happens

Next steps

Tracing

Durable runs

Streaming

Deployment

​Send runs to Celesto

​Read a run in code

​Track token usage

​See which tools ran

​Tell the three endings apart

​Failing tools do not kill the run

​Keep a copy of every run

​Watch a run as it happens

​Next steps

Tracing

Durable runs

Streaming

Deployment

Send runs to Celesto

Read a run in code

Track token usage

See which tools ran

Tell the three endings apart

Failing tools do not kill the run

Keep a copy of every run

Watch a run as it happens

Next steps