> ## Documentation Index
> Fetch the complete documentation index at: https://docs.celesto.ai/llms.txt
> Use this file to discover all available pages before exploring further.

# Monitoring agents in production

> Watch Agentor agents in production: send traces to Celesto, read token usage and tool history off a run result, and tell a real failure from a run that ran out of turns.

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](/agentor/tracing) 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:

```bash theme={null}
export CELESTO_API_KEY="cel_..."
```

```python theme={null}
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](/agentor/tracing) for setup options, private endpoints, and what each span holds.

<Tip>
  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.
</Tip>

## Read a run in code

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

```python theme={null}
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
```

<ResponseField name="final_output" type="str | BaseModel | None">
  The agent's answer. A parsed model instance when [`output_type`](/agentor/structured-output) is set, `None` if the run did not finish.
</ResponseField>

<ResponseField name="status" type="'completed' | 'max_turns' | 'failed'">
  How the run ended. Always check this before trusting `final_output`.
</ResponseField>

<ResponseField name="error" type="str | None">
  Why the run did not complete. `None` on success.
</ResponseField>

<ResponseField name="usage" type="Usage">
  `input_tokens`, `output_tokens`, and `total_tokens` summed across every model call in the run.
</ResponseField>

<ResponseField name="messages" type="list[dict]">
  The conversation as the model saw it — `user`, `assistant`, and `tool` messages. Pass it back into `arun()` to continue the conversation.
</ResponseField>

<ResponseField name="events" type="list[Event]">
  Every step: `run_start`, `generation`, `tool_call`, `tool_result`, `message`, `run_end`.
</ResponseField>

### Track token usage

```python theme={null}
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:

```python theme={null}
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:

```python theme={null}
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:

```python theme={null}
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`:

<Tabs>
  <Tab title="completed">
    The agent answered. `final_output` is set, `error` is `None`.
  </Tab>

  <Tab title="max_turns">
    The agent used its whole turn budget without settling on an answer — usually a loop between tools, or instructions it cannot satisfy.

    ```python theme={null}
    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.
  </Tab>

  <Tab title="failed">
    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.
  </Tab>
</Tabs>

```python theme={null}
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:

```python theme={null}
@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:

```python theme={null}
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
```

<Note>
  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.
</Note>

## Keep a copy of every run

Tracing is for looking at runs. A [store](/agentor/durable-runs) is for keeping them — and for finishing them if the process dies:

```python theme={null}
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:

```python theme={null}
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](/agentor/guides/streaming) for the full event shape.

## Next steps

<CardGroup cols={2}>
  <Card title="Tracing" icon="chart-line" href="/agentor/tracing">
    Set up Celesto tracing and learn what each span records.
  </Card>

  <Card title="Durable runs" icon="floppy-disk" href="/agentor/durable-runs">
    Save runs to disk and resume them after a crash.
  </Card>

  <Card title="Streaming" icon="bolt" href="/agentor/guides/streaming">
    Show progress to users while the agent works.
  </Card>

  <Card title="Deployment" icon="cloud" href="/agentor/deploy">
    Ship the agent, with observability already on.
  </Card>
</CardGroup>