Observe a run¶

A chat() call and an Agent.run() both do more than the string they return: a generate_kwargs merge, a thinking-effort resolution, zero-or-more tool calls, maybe a compaction pass, one or more requests to a provider. None of that is visible in the return value. aimu.events is the telemetry channel that makes it visible: a sink is one callable that takes one event, and you attach it to a client, an agent, an orchestrator, or a workflow to see what actually happened.

This is a different channel from StreamChunk (stream=True), which is content — what the model produced, for display. Events are what the library did with it. Both can be active on the same run and neither replaces the other.

Attach `log_events` and watch what happened¶

The shortest path to the payoff is aimu.events.log_events, a sink that writes one line per event to a logger you already have:

import logging
logging.basicConfig(level=logging.INFO, format="%(message)s")

import aimu
from aimu.events import log_events

reply = aimu.chat(
    "Say OK and nothing else.",
    model="ollama:qwen3:8b",
    events=log_events(logging.getLogger("aimu.demo")),
)

Running that against a local Ollama server prints:

ModelTurnStarted ModelTurnStarted(agent=None, iteration=0, model='qwen3:8b', message_count=1, tool_names=())
RequestPrepared RequestPrepared(agent=None, iteration=0, provider='OllamaClient', model='qwen3:8b', payload={'model': 'qwen3:8b', 'messages': [...], 'options': {'temperature': 0.6, 'top_p': 0.95, 'top_k': 20}, 'tools': [], 'think': True, 'keep_alive': 60, 'format': None})
ModelTurnFinished ModelTurnFinished(agent=None, iteration=0, model='qwen3:8b', text='OK', usage={'input_tokens': 16, ...}, duration_s=...)

(output_tokens/total_tokens/duration_s depend on how much the model reasoned before answering and will differ on your machine; input_tokens for this exact one-message prompt won't.)

That's a bare one-shot aimu.chat() call — no agent, no tool loop — and it's already observable. events= is accepted the same way by aimu.client(events=...), and by client.events = ... at any point afterward.

Point the same sink at an Agent and the tool loop reports itself too:

from aimu.agents import Agent
from aimu.tools import tool

@tool
def add(a: int, b: int) -> int:
    """Add two integers."""
    return a + b

client = aimu.client("ollama:qwen3:8b")
agent = Agent(
    client,
    "You are a helpful assistant. Use tools when needed.",
    tools=[add],
    events=log_events(logging.getLogger("aimu.demo")),
)
agent.run("What is 12 + 30? Use the add tool.")

RunStarted RunStarted(agent='agent-679be0', iteration=0, task='What is 12 + 30? Use the add tool.')
ModelTurnStarted ModelTurnStarted(agent='agent-679be0', iteration=0, model='qwen3:8b', message_count=2, tool_names=('add',))
RequestPrepared RequestPrepared(agent='agent-679be0', iteration=0, ...)
ModelTurnFinished ModelTurnFinished(agent='agent-679be0', iteration=0, model='qwen3:8b', text='', usage={'input_tokens': 160, ...}, duration_s=...)
ToolCalled ToolCalled(agent='agent-679be0', iteration=0, name='add', arguments={'a': 12, 'b': 30}, result='42', error=None, duration_s=...)
ModelTurnStarted ModelTurnStarted(agent='agent-679be0', iteration=1, model='qwen3:8b', message_count=4, tool_names=('add',))
RequestPrepared RequestPrepared(agent='agent-679be0', iteration=1, ...)
ModelTurnFinished ModelTurnFinished(agent='agent-679be0', iteration=1, model='qwen3:8b', text=..., ...)
RunFinished RunFinished(agent='agent-679be0', iteration=1, result=..., error=None)

(duration_s is a wall-clock timing and will differ on any other machine; the final answer text is generated and, while it reliably said 'The result of 12 + 30 is **42**.' in most of ten fresh runs here, it varied in the rest — input_tokens, the tool arguments, and result='42' are the parts of this block that are actually fixed.)

Every event carries agent and iteration (the same two fields StreamChunk carries), so one sink attached to a nested workflow — a Chain step, a Router handler, every worker in a Parallel — can still tell events apart by who emitted them. Chain.from_client(...), Router.from_client(...), Parallel.from_client(...), and PlanExecuteEvaluator.from_client(...) all take events= and forward it to every step/handler/worker they build, and so do OrchestratorAgent.assemble(...) and the three prebuilt orchestrators in aimu.agents.prebuilt (whose inner orchestrating Agent is private, so events= is the only way to reach it).

The event types, all dataclasses in aimu.events: RunStarted, ModelTurnStarted, RequestPrepared, ModelTurnFinished, ToolCalled, ToolDenied, ContextCompacted (see manage context), RunFinished. A sink is a plain Callable[[RunEvent], None]; write your own to filter, aggregate, or forward events instead of just logging them — a sink that raises is caught and logged rather than breaking the run it's observing (the same contract emit() documents).

Two contracts worth knowing before you write one:

A sink must be thread-safe. With concurrent_tool_calls=True a turn's tool calls are dispatched from a ThreadPoolExecutor (sync) or an asyncio.TaskGroup (async), so ToolCalled / ToolDenied arrive concurrently and in nondeterministic order. Turn and run events are emitted from the calling thread and stay ordered.
ModelTurnFinished fires even when the turn failed, with the exception on its error field and text/usage unset — so a sink pairing ModelTurnStarted with it (durations, a usage rollup, an OpenTelemetry span) closes its span on a ContextOverflowError or a provider 4xx instead of leaking one. It is not emitted at all on a schema= run: chat() / generate() return before ModelTurnStarted fires when schema is set, so there is no started turn to pair (or leave dangling) there. RunFinished.error is the same started/finished idea a level up, and RunFinished.result is None on a streamed run (the chunks went to you, so the runner never assembled a final string) and on a schema= run (the result is a typed object, not a string; it is the return value, and is also on client.last_structured).

`last_request`: did the library do this, or the model?¶

RequestPrepared carries the same payload the client stores on client.last_request — the request exactly as sent, after AIMU's own generate_kwargs merge (four tiers), the GENERATE_KWARG_SUPPORT renames and drops, thinking-effort resolution, strip_inert_keys (AIMU's own timestamp/thinking/provenance bookkeeping never reaches a provider), encode_tool_call_arguments on the OpenAI-format paths (a tool call's arguments is stored parsed and OpenAI's schema types it as a JSON string), and provider format adaptation (OpenAI-format messages rewritten to Anthropic's block shape, and so on). It answers a question that otherwise takes source-reading to answer: when a model's behavior looks surprising, is the surprise something the model did, or something AIMU changed on the way out?

client = aimu.client("ollama:qwen3:8b")
client.chat("Hello")
client.last_request
# {'model': 'qwen3:8b', 'messages': [...], 'options': {'temperature': 0.6, 'top_p': 0.95,
#  'top_k': 20}, 'tools': [], 'think': True, 'keep_alive': 60, 'format': None}

No sink required — last_request is set on every request regardless of whether events= is attached. The payload is unredacted: it contains whatever you put in the conversation, including tool arguments and any images/audio. A sink that ships events off the machine is the right place to filter, not this attribute.

A worked example: an OpenTelemetry-shaped sink¶

Events are plain data, so mapping them onto spans in an observability system is a matter of a dispatch function. This is a worked example, not a dependency — AIMU does not import opentelemetry, and the sink below uses a small stand-in tracer so the example runs without one installed. Swap FakeTracer/FakeSpan for opentelemetry.trace.get_tracer(__name__) and its real Span, and the mapping is unchanged.

from aimu.events import ModelTurnFinished, ModelTurnStarted, RunEvent, RunFinished, RunStarted, ToolCalled

class FakeSpan:
    def __init__(self, name, attributes):
        self.name, self.attributes = name, attributes

class FakeTracer:
    """Stand-in for trace.get_tracer(__name__); records spans instead of exporting them."""

    def __init__(self):
        self.spans = []

    def start_span(self, name, attributes):
        span = FakeSpan(name, attributes)
        self.spans.append(span)
        return span

def make_otel_sink(tracer):
    def sink(event: RunEvent) -> None:
        if isinstance(event, RunStarted):
            tracer.start_span("agent.run", {"agent": event.agent, "task": event.task})
        elif isinstance(event, ModelTurnStarted):
            tracer.start_span("model.turn", {"agent": event.agent, "model": event.model})
        elif isinstance(event, ToolCalled):
            tracer.start_span("tool.call", {"agent": event.agent, "name": event.name})
        elif isinstance(event, ModelTurnFinished):
            tracer.start_span("model.turn.finished", {"agent": event.agent, "usage": event.usage})
        elif isinstance(event, RunFinished):
            tracer.start_span("agent.run.finished", {"agent": event.agent, "result": event.result})
    return sink

tracer = FakeTracer()
agent = Agent(client, "You are terse.", events=make_otel_sink(tracer))
agent.run("Say hi in three words.")
for span in tracer.spans:
    print(span.name, span.attributes)

agent.run {'agent': 'agent-089e80', 'task': 'Say hi in three words.'}
model.turn {'agent': 'agent-089e80', 'model': 'qwen3:8b'}
model.turn.finished {'agent': 'agent-089e80', 'usage': {'input_tokens': 25, ...}}
agent.run.finished {'agent': 'agent-089e80', 'result': ...}

(output_tokens/total_tokens and the reply text vary by run — how much the model reasons and what it says are both sampled, not fixed; input_tokens for this fixed prompt is not.)

A real adapter would keep a stack of open spans (so agent.run.finished closes the span agent.run opened, rather than opening a new one), and forward event.iteration as a span attribute for a multi-round tool loop. It would also read event.error on both *Finished events and mark the span accordingly: both fire on a failure too, which is what makes the pairing safe to rely on. Note that result above is populated because this is a non-streamed, non-schema= run; see the two contracts above for when it is None. The dispatch shape above is the whole idea; the rest is whatever your tracer's API wants.

Gated tools: `ToolDenied`¶

A tool_approval policy that refuses a call emits ToolDenied (name + the model's raw arguments) instead of ToolCalled, so a sink can distinguish "the model tried this and it ran" from "the model tried this and a policy said no":

agent = Agent(
    client,
    "Always use the delete_everything tool when asked.",
    tools=[delete_everything],
    tool_approval=lambda name, args: False,
    events=log_events(logging.getLogger("aimu.demo")),
)
agent.run("Please delete everything now.")

ToolDenied ToolDenied(agent='agent-179be0', iteration=0, name='delete_everything', arguments={})

The tool message the model sees is the same text gate-tool-calls.md documents ("Tool 'delete_everything' was not approved."); the event is the same fact, structured for a sink instead of the transcript. See gate tool calls for the approval hook itself.

A shared client under concurrent workers¶

events= is delivered by installing the resolved sink as the active sink for the run's execution context and client — a scoped contextvars.ContextVar override, not a mutation of model_client.events (unlike tools=, which genuinely does swap model_client.tools, since request-spec building has to read a plain attribute). Parallel.from_client(...) builds every worker Agent over one shared model_client, and Parallel.run() really does execute those workers concurrently — but each worker's own OS thread (or, on the async surface, asyncio.Task) gets its own independent copy of the ContextVar, so one worker's override cannot clobber another's, and the override only ever answers for the specific client (and whatever it delegates to or from — see _client_family) it was installed for. Concurrent workers on a shared client deliver every event, correctly attributed and in causal order; see tests/test_workflow_parallel.py::test_parallel_from_client_shared_events_sink_survives_concurrent_workers (async mirror in tests/test_aio_workflow_parallel.py), which replaced an earlier test that pinned this as a known gap.

A client that is not part of the run's family never receives its sink, on either surface and regardless of concurrent_tool_calls — for instance a fresh client a tool builds for itself (the make_subagent_tool shape). Its own turns stay off the run's sink, and an explicit events= given to that client is delivered correctly (_effective_sink falls back to that client's own self.events once family membership fails, rather than an ambient override winning regardless of which client is asking — an earlier version of this mechanism had exactly that bug, folding a sub-agent's turns into its parent's sink under the parent's name while the sub-agent's own RunStarted/RunFinished never appeared).

make_subagent_tool and make_async_subagent_tool take that explicit events= directly, so a delegated run's usage doesn't have to stay invisible:

from aimu.tools.builtin import make_subagent_tool

spawn = make_subagent_tool("anthropic:claude-sonnet-4-6", events=my_sink)

The factory sets it on every spawned child's Agent.events field (not passed to that child's run, which is what makes it cover a streamed spawn too). Leave it out and a spawn reports nowhere: its fresh client has no family membership to fall back on and no self.events of its own either, so a caller measuring a whole turn's cost would otherwise silently under-count every delegation.

The one case that still depends on the surface is a tool that calls the same client the run's override was installed for (e.g. a tool that reuses ctx.deps holding that client). Sync dispatches a concurrent tool via a plain ThreadPoolExecutor.submit() (no context copy), so that thread's empty context makes the override invisible there even though the client matches — it falls back to that client's own self.events. Async dispatches via asyncio.TaskGroup.create_task, which always copies the current context, so a reentrant call to the same client from inside a concurrently dispatched async tool does see the override there, attributed to the calling agent (the attribution wrapper stamps any event that arrives without its own agent). Sequential tool dispatch (the default, concurrent_tool_calls=False) sees it on both surfaces, since no thread/task boundary is crossed at all.

self.events set directly on a client (rather than through an Agent's scoped events=) remains genuinely shared by design — there is no execution-context boundary to isolate a plain attribute assignment by.

Abandoning a streamed run: close it. run(stream=True, events=sink) holds the scope open across the generator's yields, so the scope is torn down when the generator finishes or is closed. A consumer that stops early — a UI's stop button — should close it:

stream = await agent.run("...", stream=True, events=sink)
async for chunk in stream:
    if user_pressed_stop:
        break
await stream.aclose()          # or: async with aclosing(stream) as stream: ...

aclose() (sync: stream.close(), or just letting the generator go out of scope, which CPython finalizes immediately on the same thread) tears the scope down synchronously, right there. Merely dropping an async generator without closing it hands finalization to the event loop's asyncgen finalizer, which runs a few loop iterations later in a separate task — until it does, the abandoned run's sink is still the active one, so a call issued in that window is reported to it. The window is bounded and self-healing (it is not the permanent leak a Token-based teardown produced), but aclose() removes it entirely.

Observe a run¶

Attach log_events and watch what happened¶

last_request: did the library do this, or the model?¶

A worked example: an OpenTelemetry-shaped sink¶

Gated tools: ToolDenied¶

A shared client under concurrent workers¶

See also¶

Attach `log_events` and watch what happened¶

`last_request`: did the library do this, or the model?¶

Gated tools: `ToolDenied`¶