AgentRunContext#

Understand request ownership, phases, and runtime limits. See the guide.

class zett_agent.agent.AgentRunContext(config, state, tools, extensions=(), model=None, metadata=<factory>, tags=<factory>, input_message=None, server_tools=<factory>, event_queue=<factory>, _pending_message_events=<factory>)[source]#

Per-run references shared by all lifecycle hooks.

Register request-scoped local and server tools during on_tool(). Add or replace context messages through add_message() and replace_messages(); mutate either tool registry in place to update the current request without leaking registrations into another request or session.

Note

Context identity is request-scoped. It is safe to use a context as a key for temporary extension state only when every success, error, and cancellation path removes that entry.

Examples

Register a tool for only the current request:

async def on_tool(self, context: AgentRunContext) -> None:
    context.register_tool(read_file)

Register a provider-hosted tool without adding a local executor:

async def on_tool(self, context: AgentRunContext) -> None:
    context.register_server_tool(
        ServerToolDefinition(
            type="web_search",
            configuration={"search_context_size": "medium"},
        )
    )

Publish an internal notification to every extension:

await context.publish(MyIndexReadyEvent(document_count=12))

Or deliver it to one named extension:

await context.publish(event, target="SearchIndexExtension")

See also

AgentExtension for lifecycle hooks and AgentTool for request-scoped tool registration.

Parameters:
property session_id: str#

The session this run belongs to.

config.session_id is optional because a caller may leave it out, and the runtime fills it in before a context exists, so by the time an extension holds a context this is always set. Use it in place of config.session_id wherever a str is needed, such as a dictionary key.

async emit(event)[source]#

Queue one extension-owned CUSTOM or compaction event for the caller.

Parameters:

event (AgentEvent)

Return type:

None

register_tool(tool)[source]#

Register one request-scoped tool while rejecting ambiguous names.

Parameters:

tool (AgentTool)

Return type:

None

register_server_tool(tool)[source]#

Register one request-scoped provider tool by its unique wire type.

The opaque configuration is copied because extensions may reuse one definition across concurrent sessions. Server tools are model-side capabilities and therefore never appear in tools, which is the registry used by the Agent’s local executor.

Raises:

ValueError – If this request already contains the same server-tool type. Registering two configurations under one discriminator would make provider behavior order-dependent.

Parameters:

tool (ServerToolDefinition)

Return type:

None

async publish(event, *, target=None)[source]#

Deliver an event to one named extension or broadcast it by default.

A target must match exactly one registered extension name. Unknown or ambiguous targets raise ValueError without delivering the event. During broadcast, handler errors propagate immediately and stop delivery. Already completed work is not rolled back. Subscribers may retain events.

Examples

Usage:

await context.publish(CompactionEvent(
    compressed_from=1, compressed_to=20,
    kept_from=21, kept_to=30, summary="Earlier decisions...",
))

await context.publish(event, target="MetricsExtension")
Parameters:
Return type:

None

async append_message(message, timing, usage=None)[source]#

Append one new message and notify persistence subscribers.

Runtime code must use this method for produced messages so in-memory context and persistence notifications cannot drift apart. Subscribers decide from message.persist whether the message belongs in storage.

The message is visible in state.messages before subscribers run. Subscriber failures propagate and do not roll back the in-memory append.

Parameters:
Return type:

None

add_message(message, *, index=None, usage=None)[source]#

Add one extension-owned message and queue its append event.

The queued event is flushed by the runtime after setup hooks or before the next model request. Extensions must use this method instead of mutating AgentState.messages directly.

Parameters:
Return type:

None

replace_messages(messages, *, emit_new=True)[source]#

Replace model context and optionally queue events for newly added messages.

Restored history and compaction use emit_new=False because those messages are already represented by storage records or a snapshot.

Parameters:
Return type:

None

add_message_event(message, usage=None)[source]#

Queue one append event for a message already present in state.

Parameters:
Return type:

None

async flush_message_events()[source]#

Publish every queued message event exactly once.

Return type:

None