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>)[源代码]#

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.

备注

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.

示例

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")

参见

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

参数:
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)[源代码]#

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

参数:

event (AgentEvent)

返回类型:

None

register_tool(tool)[源代码]#

Register one request-scoped tool while rejecting ambiguous names.

参数:

tool (AgentTool)

返回类型:

None

register_server_tool(tool)[源代码]#

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.

抛出:

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

参数:

tool (ServerToolDefinition)

返回类型:

None

async publish(event, *, target=None)[源代码]#

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.

示例

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")
参数:
返回类型:

None

async append_message(message, timing, usage=None)[源代码]#

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.

参数:
返回类型:

None

add_message(message, *, index=None, usage=None)[源代码]#

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.

参数:
返回类型:

None

replace_messages(messages, *, emit_new=True)[源代码]#

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.

参数:
返回类型:

None

add_message_event(message, usage=None)[源代码]#

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

参数:
返回类型:

None

async flush_message_events()[源代码]#

Publish every queued message event exactly once.

返回类型:

None