"""Optional creation and event-handling conveniences around the unchanged Agent."""
from collections.abc import AsyncIterator, Mapping, Sequence
from ._agent.lifecycle import final_answer
from ._compat import aclosing_stream
from .agent import Agent, AgentRunConfig
from .dispatcher import AgentEventDispatcher
from .events import AgentEvent
from .extensions.base import AgentExtension
from .extensions.events import CompactionEvent
from .json_types import JsonValue
from .messages import AssistantMessage, UserMessage
from .model import AgentModel, ReasoningEffort
from .resume import ResumePoint
from .sync_runtime import SyncMethodsMixin
from .tools.base import AgentTool
[docs]
class AgentClient(SyncMethodsMixin):
"""Compose an initialized Agent with optional application event callbacks.
The public agent attribute exposes the original runtime for advanced APIs,
including emit_external_event(). Calling it directly bypasses this client's
callbacks. No methods or attributes are injected into the original Agent.
Callback failures propagate to the caller and close the underlying stream:
unfinished requests are cancelled through the Agent's normal close path.
If a callback fails after completion, the request remains completed.
Shared handlers must distinguish sessions using event.session_id.
Examples:
Usage::
agent = await Agent.create(model, config=AgentRunConfig(session_id="existing"))
client = AgentClient(agent, event_dispatcher=ConsoleEvents())
reply = await client.run("Continue")
"""
def __init__(self, agent: Agent, *, event_dispatcher: AgentEventDispatcher | None = None) -> None:
self.agent = agent
self.event_dispatcher = event_dispatcher
[docs]
async def stream(
self,
message: UserMessage | str,
*,
config: AgentRunConfig | None = None,
model: AgentModel | None = None,
reasoning_effort: ReasoningEffort | None = None,
parallel_tool_call: bool | None = None,
metadata: Mapping[str, JsonValue] | None = None,
tags: Mapping[str, JsonValue] | None = None,
) -> AsyncIterator[AgentEvent]:
"""Dispatch then yield each original event, in order, exactly once.
Use aclosing(client.stream(...)) when stopping consumption early. Do
not dispatch yielded events manually if a handler is already bound.
Args:
message: User text or an explicit multimodal UserMessage.
config: Optional per-request session and request identity.
model: Request-specific model; omitted to use the Agent default.
reasoning_effort: Per-request override of the model reasoning level.
parallel_tool_call: Per-request override of parallel tool calling.
metadata: Request-scoped JSON data supplied to extensions.
tags: Request-scoped classifications supplied to extensions.
Yields:
Original AgentEvent objects after the bound callback completes.
Note:
Closing early cancels unfinished work. Provider, callback, and
cancellation exceptions are not converted into synthetic events.
"""
stream = self.agent.stream(
message,
config=config,
model=model,
reasoning_effort=reasoning_effort,
parallel_tool_call=parallel_tool_call,
metadata=metadata,
tags=tags,
)
async with aclosing_stream(self._dispatched(stream)) as events:
async for event in events:
yield event
async def _dispatched(self, events: AsyncIterator[AgentEvent]) -> AsyncIterator[AgentEvent]:
"""Hand each event to the bound dispatcher, if there is one, then yield it unchanged."""
async with aclosing_stream(events) as stream:
async for event in stream:
if self.event_dispatcher is not None:
await self.event_dispatcher.dispatch(event)
yield event
[docs]
async def compact(
self,
*,
config: AgentRunConfig | None = None,
model: AgentModel | None = None,
metadata: Mapping[str, JsonValue] | None = None,
tags: Mapping[str, JsonValue] | None = None,
) -> CompactionEvent | None:
"""Compact the session's history now, dispatching every event it emits.
This is the one-call form of :meth:`Agent.compact`: the pass runs to
completion, each event reaches the bound dispatcher in order — so a UI
shows the same progress a streamed run shows — and nothing is appended
to the conversation or sent to the primary model.
Args:
config: Session identity; defaults to the initialized configuration.
model: Model an extension compacts with when it has none of its own.
metadata: Request data visible to extensions and persistence.
tags: Request classifications visible to extensions and persistence.
Returns:
The checkpoint the pass stored, or None when the extensions found
nothing worth replacing.
Examples:
Usage::
stored = await client.compact()
"""
checkpoint: CompactionEvent | None = None
async with aclosing_stream(
self.agent.compact(config=config, model=model, metadata=metadata, tags=tags)
) as events:
async for event in events:
if isinstance(event.compaction, CompactionEvent):
checkpoint = event.compaction
if self.event_dispatcher is not None:
await self.event_dispatcher.dispatch(event)
return checkpoint
[docs]
async def run(
self,
message: UserMessage | str,
*,
config: AgentRunConfig | None = None,
model: AgentModel | None = None,
reasoning_effort: ReasoningEffort | None = None,
parallel_tool_call: bool | None = None,
metadata: Mapping[str, JsonValue] | None = None,
tags: Mapping[str, JsonValue] | None = None,
) -> AssistantMessage:
"""Collect a complete stream with callbacks and return its final answer.
Args:
message: User text or an explicit multimodal UserMessage.
config: Per-request identity; omitted to reuse the initialized session.
model: Request-specific model; omitted to use the Agent default.
reasoning_effort: Override the runtime default for this request only.
parallel_tool_call: Override parallel tool calling for this request.
metadata: JSON-compatible application data for persistence/extensions.
tags: JSON-compatible classifications for persistence/extensions.
Returns:
The final AssistantMessage after the stream closes successfully.
Raises:
AgentProtocolError: If no final answer is emitted or the runtime is
not initialized. Model and callback failures propagate unchanged.
"""
return await final_answer(
self.stream(
message,
config=config,
model=model,
reasoning_effort=reasoning_effort,
parallel_tool_call=parallel_tool_call,
metadata=metadata,
tags=tags,
)
)
[docs]
async def resume_point(
self,
*,
config: AgentRunConfig | None = None,
model: AgentModel | None = None,
) -> ResumePoint | None:
"""Report where the session was interrupted, or None when there is nothing to resume.
See :meth:`Agent.resume_point`. Nothing is written, no model is called, and no event
reaches the bound dispatcher.
"""
return await self.agent.resume_point(config=config, model=model)
[docs]
async def stream_resume(
self,
*,
config: AgentRunConfig | None = None,
model: AgentModel | None = None,
reasoning_effort: ReasoningEffort | None = None,
parallel_tool_call: bool | None = None,
metadata: Mapping[str, JsonValue] | None = None,
tags: Mapping[str, JsonValue] | None = None,
) -> AsyncIterator[AgentEvent]:
"""Dispatch then yield each event of :meth:`Agent.stream_resume`, in order, exactly once.
Use aclosing(client.stream_resume(...)) when stopping consumption early. A session with
nothing to resume raises :class:`~zett_agent.exceptions.AgentResumeError`.
"""
stream = self.agent.stream_resume(
config=config,
model=model,
reasoning_effort=reasoning_effort,
parallel_tool_call=parallel_tool_call,
metadata=metadata,
tags=tags,
)
async with aclosing_stream(self._dispatched(stream)) as events:
async for event in events:
yield event
[docs]
async def resume(
self,
*,
config: AgentRunConfig | None = None,
model: AgentModel | None = None,
reasoning_effort: ReasoningEffort | None = None,
parallel_tool_call: bool | None = None,
metadata: Mapping[str, JsonValue] | None = None,
tags: Mapping[str, JsonValue] | None = None,
) -> AssistantMessage:
"""Pick an interrupted session up where it stopped, with callbacks, and return the answer.
See :meth:`Agent.resume` for what is done in each case and what it raises.
"""
return await final_answer(
self.stream_resume(
config=config,
model=model,
reasoning_effort=reasoning_effort,
parallel_tool_call=parallel_tool_call,
metadata=metadata,
tags=tags,
)
)
[docs]
async def create_agent(
model: AgentModel | None,
*,
config: AgentRunConfig | None = None,
system_prompt: str = "You are a helpful assistant.",
tools: Sequence[AgentTool] = (),
extensions: Sequence[AgentExtension] | None = None,
reasoning_effort: ReasoningEffort = ReasoningEffort.MEDIUM,
parallel_tool_call: bool = True,
max_iterations: int = 36,
max_internal_messages: int = 8,
event_dispatcher: AgentEventDispatcher | None = None,
) -> AgentClient:
"""Create an initialized runtime wrapped in an event-aware AgentClient.
Omit config for a new UUIDv7 session, or pass an existing session ID to
restore history through your configured persistence extension. All runtime
options pass through unchanged; extensions=None keeps Agent defaults.
The supplied model and extension resources remain caller-owned.
Args:
model: Default provider adapter, or None when each request supplies one.
config: Stable session identity; omitted to generate a new UUIDv7 ID.
system_prompt: Initial instructions rebuilt before each request.
tools: Explicit typed tools available to the runtime.
extensions: Lifecycle and middleware extensions. None keeps default
memory and tool guidance; an explicit sequence replaces those defaults.
reasoning_effort: Default reasoning level, overridable per request.
parallel_tool_call: Allow supported providers and local tools to use
parallel calls. Individual tools may explicitly require serial execution.
max_iterations: Maximum model calls for each user or internal message.
max_internal_messages: Maximum internal continuations per request.
event_dispatcher: Optional callbacks awaited in stream order.
Returns:
An initialized AgentClient wrapping an ordinary Agent.
Raises:
ValueError: If runtime limits or extension/tool identities are invalid.
Examples:
Usage::
from zett_agent.events import AgentEvent
from zett_agent.dispatcher import AgentEventDispatcher
from zett_agent.client import create_agent
class ConsoleEvents(AgentEventDispatcher):
async def on_text_delta_event(self, event: AgentEvent) -> None:
print(event.delta, end="", flush=True)
client = await create_agent(model, event_dispatcher=ConsoleEvents())
reply = await client.run("Hello")
await client.run("Continue in the same session")
"""
agent = await Agent.create(
model,
config=config,
system_prompt=system_prompt,
tools=tools,
extensions=extensions,
reasoning_effort=reasoning_effort,
parallel_tool_call=parallel_tool_call,
max_iterations=max_iterations,
max_internal_messages=max_internal_messages,
)
return AgentClient(agent, event_dispatcher=event_dispatcher)
