Agent API Reference¶

The main Agent. Can do all sort of crazy things.

Agent ¶

Bases: BaseAgent[TDeps, OutputDataT]

The main agent class.

Generically typed with: Agent[Type of Dependencies, Type of Result]

agent_pool `property` `writable` ¶

agent_pool: AgentPool[Any] | None

Compatibility shim: returns the pool reference.

.. deprecated:: M2 Use :attr:host_context instead. The agent_pool property will be removed in M3. host_context provides the same infrastructure fields via an immutable :class:HostContext.

agent_type `property` ¶

agent_type: str

Return the agent-type string used for persistence.

This is the persistence-domain identifier (stored in SessionData.agent_type). It differs from :data:SourceType which is the event-domain identifier used during streaming.

Subclasses may override this to provide a more specific type string (e.g. "native", "acp", "claude_code").

Returns:

Type	Description
`str`	A string identifying the agent type for storage purposes.

command_store `property` ¶

command_store: CommandStore

Get the command store for slash commands.

connection_stats `property` ¶

connection_stats: AggregatedTalkStats

Get stats for all active connections of this node.

display_name `property` ¶

display_name: str

Get human-readable display name, falls back to name.

host_context `property` ¶

host_context: HostContext | None

Return HostContext from the pool, if available.

internal_fs `property` ¶

internal_fs: IsolatedMemoryFileSystem

Get the internal filesystem for tool/session state.

Tools can use this to store logs, history, temporary files, etc. Access via AgentContext.fs in tool implementations.

Returns:

Type	Description
`IsolatedMemoryFileSystem`	In-memory filesystem scoped to this agent

lifecycle_config `property` ¶

lifecycle_config: LifecycleConfig | None

Lifecycle configuration for this agent's RunLoop.

Returns:

Type	Description
`LifecycleConfig \| None`	The LifecycleConfig if set, or `None` for all-defaults.

lifecycle_dimensions `property` ¶

lifecycle_dimensions: tuple[Any, Any, Any, Any, Any] | None

Pre-created lifecycle dimensions for the current run, if any.

Returns:

Type	Description
`tuple[Any, Any, Any, Any, Any] \| None`	Tuple of ``(trigger_source, journal, snapshot_store,
`tuple[Any, Any, Any, Any, Any] \| None`	comm_channel, event_transport)`` if dimensions were created
`tuple[Any, Any, Any, Any, Any] \| None`	from `lifecycle_config`, or `None` if defaults should be
`tuple[Any, Any, Any, Any, Any] \| None`	used.

message_received `class-attribute` `instance-attribute` ¶

message_received = Signal[ChatMessage[Any]]()

Signal emitted when node receives a message.

message_sent `class-attribute` `instance-attribute` ¶

message_sent = Signal[ChatMessage[Any]]()

Signal emitted when node creates a message.

model_name `property` ¶

model_name: str | None

Get the model name in a consistent format (provider:model_name).

name `property` `writable` ¶

name: str

Get agent name.

overlay_fs `property` ¶

overlay_fs: OverlayFileSystem

Get unified filesystem view combining agent storage and VFS resources.

Provides a layered filesystem where: - Writes go to the agent's internal filesystem (upper layer) - Reads fall through to VFS resources if not found locally

Returns:

Type	Description
`OverlayFileSystem`	OverlayFileSystem combining internal_fs and pool's VFS registry

session_id `property` `writable` ¶

session_id: str | None

Current conversation session bound to this agent, if any.

storage `property` ¶

storage: StorageManager | None

Get storage manager from pool.

task_manager `property` ¶

task_manager: _TaskManagerShim

Deprecated: backward-compat shim for the old TaskManager API.

.. deprecated:: Use :meth:spawn_task or :attr:_pending_tasks directly.

Returns:

Type	Description
`_TaskManagerShim`	A shim object that provides the old TaskManager method surface
`_TaskManagerShim`	(`complete_tasks`, `cleanup_tasks`, `fire_and_forget`,
`_TaskManagerShim`	`create_task`, `_pending_tasks`) backed by this node's
`_TaskManagerShim`	`_pending_tasks` set.

AgentReset `dataclass` ¶

Emitted when agent is reset.

InterruptEvent `dataclass` ¶

Emitted when agent is interrupted.

aenter `async` ¶

__aenter__() -> Self

Enter async context and set up MCP servers.

aexit `async` ¶

__aexit__(
    exc_type: type[BaseException] | None,
    exc_val: BaseException | None,
    exc_tb: TracebackType | None,
) -> None

Exit async context.

and ¶

__and__(
    other: ProcessorCallback[Any] | BaseTeam[TDeps, Any] | Agent[TDeps, Any],
) -> BaseTeam[TDeps, Any]

__and__(other: ProcessorCallback[Any] | BaseTeam[Any, Any] | Agent[Any, Any]) -> BaseTeam[Any, Any]

__and__(other: MessageNode[Any, Any] | ProcessorCallback[Any]) -> BaseTeam[Any, Any]

Create parallel team using & operator.

Example

group = analyzer & planner & executor # Create group of 3 group = analyzer & existing_group # Add to existing group

init ¶

__init__(
    name: str = "agentpool",
    *,
    deps_type: type[TDeps] | None = None,
    model: ModelType,
    output_type: OutputSpec[OutputDataT] = str,
    session: SessionIdType | SessionQuery | MemoryConfig | bool = None,
    system_prompt: AnyPromptType | Sequence[AnyPromptType] = (),
    description: str | None = None,
    display_name: str | None = None,
    tools: Sequence[ToolType] | None = None,
    toolsets: Sequence[AbstractCapability] | None = None,
    mcp_servers: Sequence[str | MCPServerConfig] | None = None,
    resources: Sequence[PromptType | str] = (),
    skills_paths: Sequence[JoinablePathLike] | None = None,
    retries: int = 1,
    output_retries: int | None = None,
    end_strategy: EndStrategy = "early",
    input_provider: InputProvider | None = None,
    parallel_init: bool = True,
    model_settings: ModelSettings | None = None,
    event_handlers: Sequence[AnyEventHandlerType] | None = None,
    agent_pool: AgentPool[Any] | None = None,
    tool_mode: ToolMode | None = None,
    knowledge: Knowledge | None = None,
    agent_config: NativeAgentConfig | None = None,
    env: ExecutionEnvironment | StrPath | None = None,
    hooks: AgentHooks | None = None,
    tool_confirmation_mode: ToolConfirmationMode = "per_tool",
    builtin_tools: Sequence[AgentNativeTool] | None = None,
    usage_limits: UsageLimits | None = None,
    providers: Sequence[ProviderType] | None = None,
    commands: Sequence[BaseCommand] | None = None,
    metadata: dict[str, Any] | None = None,
    history_processors: Sequence[Callable[..., Any]] | None = None,
    capabilities: list[Any] | None = None,
    resolved_model_config: BaseModelConfig | None = None
) -> None

Initialize agent.

Parameters:

Name	Type	Description	Default
`name`	`str`	Identifier for the agent (used for logging and lookups)	`'agentpool'`
`deps_type`	`type[TDeps] \| None`	Type of dependencies to use	`None`
`model`	`ModelType`	The default model to use (defaults to GPT-5)	required
`output_type`	`OutputSpec[OutputDataT]`	The default output type to use (defaults to str)	`str`
`context`		Agent context with configuration	required
`session`	`SessionIdType \| SessionQuery \| MemoryConfig \| bool`	Memory configuration. - None: Default memory config - False: Disable message history (max_messages=0) - int: Max tokens for memory - str/UUID: Session identifier - MemoryConfig: Full memory configuration - MemoryProvider: Custom memory provider - SessionQuery: Session query	`None`
`system_prompt`	`AnyPromptType \| Sequence[AnyPromptType]`	System prompts for the agent	`()`
`description`	`str \| None`	Description of the Agent ("what it can do")	`None`
`display_name`	`str \| None`	Human-readable display name (falls back to name)	`None`
`tools`	`Sequence[ToolType] \| None`	List of tools to register with the agent	`None`
`toolsets`	`Sequence[AbstractCapability] \| None`	List of toolset resource providers for the agent	`None`
`mcp_servers`	`Sequence[str \| MCPServerConfig] \| None`	MCP servers to connect to	`None`
`resources`	`Sequence[PromptType \| str]`	Additional resources to load	`()`
`skills_paths`	`Sequence[JoinablePathLike] \| None`	Local directories to search for agent-specific skills	`None`
`retries`	`int`	Default number of retries for failed operations	`1`
`output_retries`	`int \| None`	Max retries for result validation (defaults to retries)	`None`
`end_strategy`	`EndStrategy`	Strategy for handling tool calls that are requested alongside a final result	`'early'`
`input_provider`	`InputProvider \| None`	Provider for human input (tool confirmation / HumanProviders)	`None`
`parallel_init`	`bool`	Whether to initialize resources in parallel	`True`
`model_settings`	`ModelSettings \| None`	Settings for the AI model	`None`
`event_handlers`	`Sequence[AnyEventHandlerType] \| None`	Sequence of event handlers to register with the agent	`None`
`agent_pool`	`AgentPool[Any] \| None`	AgentPool instance for managing agent resources	`None`
`tool_mode`	`ToolMode \| None`	Tool execution mode (None or "codemode")	`None`
`knowledge`	`Knowledge \| None`	Knowledge sources for this agent	`None`
`agent_config`	`NativeAgentConfig \| None`	Agent configuration	`None`
`env`	`ExecutionEnvironment \| StrPath \| None`	Execution environment for code/command execution and filesystem access	`None`
`hooks`	`AgentHooks \| None`	AgentHooks instance for intercepting agent behavior at run and tool events	`None`
`tool_confirmation_mode`	`ToolConfirmationMode`	Tool confirmation mode	`'per_tool'`
`builtin_tools`	`Sequence[AgentNativeTool] \| None`	PydanticAI builtin tools (WebSearchTool, CodeExecutionTool, etc.)	`None`
`usage_limits`	`UsageLimits \| None`	Per-request usage limits (applied to each run() call independently, not cumulative across the session)	`None`
`providers`	`Sequence[ProviderType] \| None`	Model providers for model discovery (e.g., ["openai", "anthropic"]). Defaults to ["models.dev"] if not specified.	`None`
`commands`	`Sequence[BaseCommand] \| None`	Slash commands	`None`
`metadata`	`dict[str, Any] \| None`	Arbitrary metadata for the agent (e.g., feature flags)	`None`
`history_processors`	`Sequence[Callable[..., Any]] \| None`	Callable history processors for message processing	`None`
`capabilities`	`list[Any] \| None`	Extra capability instances or configs to attach	`None`
`resolved_model_config`	`BaseModelConfig \| None`	Resolved model config (after variant lookup). Used for capability resolution when the agent references a model variant by name.	`None`

rshift ¶

__rshift__(other: MessageNode[Any, Any] | ProcessorCallback[Any]) -> Talk[TResult]

__rshift__(other: Sequence[MessageNode[Any, Any] | ProcessorCallback[Any]]) -> TeamTalk[TResult]

__rshift__(
    other: (
        MessageNode[Any, Any]
        | ProcessorCallback[Any]
        | Sequence[MessageNode[Any, Any] | ProcessorCallback[Any]]
    ),
) -> Talk[Any] | TeamTalk[Any]

Connect agent to another agent or group.

Example

agent >> other_agent # Connect to single agent agent >> (agent2 & agent3) # Connect to group agent >> "other_agent" # Connect by name (needs pool)

connect_to ¶

connect_to(
    target: MessageNode[Any, Any] | ProcessorCallback[Any],
    *,
    queued: Literal[True],
    queue_strategy: Literal["concat"]
) -> Talk[str]

connect_to(
    target: MessageNode[Any, Any] | ProcessorCallback[Any],
    *,
    connection_type: ConnectionType = "run",
    name: str | None = None,
    priority: int = 0,
    delay: timedelta | None = None,
    queued: bool = False,
    queue_strategy: QueueStrategy = "latest",
    transform: AnyTransformFn[Any] | None = None,
    filter_condition: AsyncFilterFn | None = None,
    stop_condition: AsyncFilterFn | None = None,
    exit_condition: AsyncFilterFn | None = None
) -> Talk[TResult]

connect_to(
    target: Sequence[MessageNode[Any, Any] | ProcessorCallback[Any]],
    *,
    queued: Literal[True],
    queue_strategy: Literal["concat"]
) -> TeamTalk[str]

connect_to(
    target: Sequence[MessageNode[Any, TResult] | ProcessorCallback[TResult]],
    *,
    connection_type: ConnectionType = "run",
    name: str | None = None,
    priority: int = 0,
    delay: timedelta | None = None,
    queued: bool = False,
    queue_strategy: QueueStrategy = "latest",
    transform: AnyTransformFn[Any] | None = None,
    filter_condition: AsyncFilterFn | None = None,
    stop_condition: AsyncFilterFn | None = None,
    exit_condition: AsyncFilterFn | None = None
) -> TeamTalk[TResult]

connect_to(
    target: Sequence[MessageNode[Any, Any] | ProcessorCallback[Any]],
    *,
    connection_type: ConnectionType = "run",
    name: str | None = None,
    priority: int = 0,
    delay: timedelta | None = None,
    queued: bool = False,
    queue_strategy: QueueStrategy = "latest",
    transform: AnyTransformFn[Any] | None = None,
    filter_condition: AsyncFilterFn | None = None,
    stop_condition: AsyncFilterFn | None = None,
    exit_condition: AsyncFilterFn | None = None
) -> TeamTalk

connect_to(
    target: (
        MessageNode[Any, Any]
        | ProcessorCallback[Any]
        | Sequence[MessageNode[Any, Any] | ProcessorCallback[Any]]
    ),
    *,
    connection_type: ConnectionType = "run",
    name: str | None = None,
    priority: int = 0,
    delay: timedelta | None = None,
    queued: bool = False,
    queue_strategy: QueueStrategy = "latest",
    transform: AnyTransformFn[Any] | None = None,
    filter_condition: AsyncFilterFn | None = None,
    stop_condition: AsyncFilterFn | None = None,
    exit_condition: AsyncFilterFn | None = None
) -> Talk[Any] | TeamTalk

Create connection(s) to target(s).

create_run ¶

create_run(
    prompt: str,
    run_ctx: AgentRunContext,
    message_history: list[ModelMessage],
    event_bus: EventBus,
    session: SessionState,
) -> RunHandle

Construct a RunHandle for v2 session-level execution.

This is the v2 entry point that replaces the legacy run() method for session-managed runs. It only constructs the RunHandle — no execution happens here. The caller is responsible for calling run_handle.start(prompt) to begin the idle/wake/turn loop, or for using :meth:create_run_stream which wraps that pattern.

Parameters:

Name	Type	Description	Default
`prompt`	`str`	Initial user prompt for the first turn. Not used during construction; pass it to `start()` when ready.	required
`run_ctx`	`AgentRunContext`	Per-run isolated context.	required
`message_history`	`list[ModelMessage]`	Incoming message history for the first turn.	required
`event_bus`	`EventBus`	Event bus for publishing stream events.	required
`session`	`SessionState`	Per-session state containing the turn lock.	required

Returns:

Type	Description
`RunHandle`	A RunHandle wired with agent, event_bus, session, and
`RunHandle`	run_ctx, ready to be started via `start(prompt)`.

create_run_stream `async` ¶

create_run_stream(
    prompt: str,
    run_ctx: AgentRunContext,
    message_history: list[ModelMessage],
    event_bus: EventBus,
    session: SessionState,
) -> AsyncGenerator[RichAgentStreamEvent[TResult]]

Run agent with streaming output via the v2 RunHandle lifecycle.

This is the v2 streaming wrapper around :meth:create_run and RunHandle.start(). It constructs a RunHandle, starts the idle/wake/turn loop, yields all stream events, and closes the handle when the stream completes.

Parameters:

Name	Type	Description	Default
`prompt`	`str`	Initial user prompt for the first turn.	required
`run_ctx`	`AgentRunContext`	Per-run isolated context.	required
`message_history`	`list[ModelMessage]`	Incoming message history for the first turn.	required
`event_bus`	`EventBus`	Event bus for publishing stream events.	required
`session`	`SessionState`	Per-session state containing the turn lock.	required

Yields:

Type	Description
`AsyncGenerator[RichAgentStreamEvent[TResult]]`	Stream events from the turn execution.

create_turn ¶

create_turn(
    prompts: list[UserContent],
    run_ctx: AgentRunContext,
    message_history: list[ModelMessage],
    **pydantic_ai_kwargs: Any
) -> Turn

Create a NativeTurn for single-cycle execution.

Parameters:

Name	Type	Description	Default
`prompts`	`list[UserContent]`	Pre-converted prompt strings for this turn.	required
`run_ctx`	`AgentRunContext`	Per-run isolated context.	required
`message_history`	`list[ModelMessage]`	Incoming message history.	required
`**pydantic_ai_kwargs`	`Any`	Extra kwargs forwarded to `NativeTurn.__init__()` → `agentlet.iter()` (e.g. `deferred_tool_results` for crash recovery resume).	`{}`

Returns:

Type	Description
`Turn`	A NativeTurn instance for single-cycle execution.

disconnect_all `async` ¶

disconnect_all() -> None

Disconnect from all nodes.

emit_agent_event `async` ¶

emit_agent_event(event: RichAgentStreamEvent[Any], source_session_id: str | None = None) -> None

Emit an agent stream event via the event manager.

Parameters:

Name	Type	Description	Default
`event`	`RichAgentStreamEvent[Any]`	The agent stream event to emit	required
`source_session_id`	`str \| None`	Optional ID of the session that produced the event	`None`

ensure_initialized `async` ¶

ensure_initialized() -> None

Wait for deferred initialization to complete.

Subclasses that use deferred init should: 1. Set self._connect_pending = True in __aenter__ 2. Override this method to do actual connection work 3. Set self._connect_pending = False when done

The base implementation is a no-op for agents without deferred init.

fire_and_forget ¶

fire_and_forget(coro: Coroutine[Any, Any, Any]) -> None

Run a coroutine in the background without waiting for result.

The coroutine is wrapped in a try/except that logs and swallows exceptions, preventing one non-critical task failure from propagating.

Parameters:

Name	Type	Description	Default
`coro`	`Coroutine[Any, Any, Any]`	Coroutine to run in the background.	required

from_callback `classmethod` ¶

from_callback(
    callback: Callable[..., Awaitable[TResult]], *, name: str | None = None, **kwargs: Any
) -> Agent[None, TResult]

from_callback(
    callback: Callable[..., TResult], *, name: str | None = None, **kwargs: Any
) -> Agent[None, TResult]

from_callback(
    callback: ProcessorCallback[Any], *, name: str | None = None, **kwargs: Any
) -> Agent[None, Any]

Create an agent from a processing callback.

Parameters:

Name	Type	Description	Default
`callback`	`ProcessorCallback[Any]`	Function to process messages. Can be: - sync or async - with or without context - must return str for pipeline compatibility	required
`name`	`str \| None`	Optional name for the agent	`None`
`kwargs`	`Any`	Additional arguments for agent	`{}`

from_config `classmethod` ¶

from_config(
    config: NativeAgentConfig,
    *,
    event_handlers: Sequence[AnyEventHandlerType] | None = None,
    input_provider: InputProvider | None = None,
    agent_pool: AgentPool[Any] | None = None,
    deps_type: type[TDeps] | None = None
) -> Self

Create a native Agent from a config object.

This is the preferred way to instantiate an Agent from configuration. Handles system prompt resolution, model resolution, toolsets setup, etc.

Parameters:

Name	Type	Description	Default
`config`	`NativeAgentConfig`	Native agent configuration	required
`name`		Optional name override (used for manifest lookups, defaults to config.name)	required
`event_handlers`	`Sequence[AnyEventHandlerType] \| None`	Optional event handlers (merged with config handlers)	`None`
`input_provider`	`InputProvider \| None`	Optional input provider for user interactions	`None`
`agent_pool`	`AgentPool[Any] \| None`	Optional agent pool for coordination	`None`
`deps_type`	`type[TDeps] \| None`	Optional dependency type	`None`

Returns:

Type	Description
`Self`	Configured Agent instance

get_active_run_context ¶

get_active_run_context(session_id: str | None = None) -> AgentRunContext | None

Get the currently active run context.

Public API for external callers (e.g., SessionPool) to check if a turn is active and access the run context without relying on private attributes.

Uses three-level fallback: 1. ContextVar (_current_run_ctx_var) for the current task 2. SessionPool lookup when pooled (via session_id or agent_pool) 3. _background_run_ctx for background task state

Parameters:

Name	Type	Description	Default
`session_id`	`str \| None`	Optional session ID for SessionPool lookup. When provided, used for the SessionPool fallback instead of instance state.	`None`

Returns:

Type	Description
`AgentRunContext \| None`	The active run context, or None if no turn is running.

get_agentlet `async` ¶

get_agentlet(
    model: ModelType | None,
    output_type: type[AgentOutputType] | None,
    input_provider: InputProvider | None = None,
    run_ctx: AgentRunContext | None = None,
) -> Agent[AgentContext[TDeps], AgentOutputType]

Create pydantic-ai agent from current state.

get_available_models `async` ¶

get_available_models() -> list[ModelInfo] | None

Get available models for this agent.

Fetches model data from the npmmirror CDN (Alibaba China mirror), which mirrors the @opencode-ai/models npm package containing the same data as models.dev/api.json. This avoids direct access to models.dev which may be unreachable from China.

Falls back to the original tokonomics discovery when the MODELS_DEV_FALLBACK environment variable is set to "1".

Returns:

Type	Description
`list[ModelInfo] \| None`	List of tokonomics ModelInfo, or None if discovery fails

get_context ¶

get_context(
    data: Any = None,
    input_provider: InputProvider | None = None,
    tool_call_id: str | None = None,
    tool_input: dict[str, Any] | None = None,
    tool_name: str | None = None,
    run_ctx: AgentRunContext | None = None,
) -> AgentContext[Any]

Create a new context for this agent.

Parameters:

Name	Type	Description	Default
`data`	`Any`	Optional custom data to attach to the context	`None`
`input_provider`	`InputProvider \| None`	Optional input provider override	`None`
`tool_call_id`	`str \| None`	Optional tool call ID	`None`
`tool_input`	`dict[str, Any] \| None`	Optional tool input	`None`
`tool_name`	`str \| None`	Optional tool name	`None`
`run_ctx`	`AgentRunContext \| None`	Optional per-run context for accessing run-isolated state	`None`

Returns:

Type	Description
`AgentContext[Any]`	A new AgentContext instance

get_mcp_server_info `async` ¶

get_mcp_server_info() -> dict[str, MCPServerStatus]

Get information about configured MCP servers.

Returns a dict mapping server names to their status info. Used by the OpenCode /mcp endpoint to display MCP servers in the UI.

The default implementation checks external capabilities for MCP servers. Subclasses may override to provide agent-specific MCP server info.

Returns:

Type	Description
`dict[str, MCPServerStatus]`	Dict mapping server name to MCPServerStatus

get_message_history `async` ¶

get_message_history(
    session_id: str | None = None, limit: int | None = None
) -> list[ChatMessage[Any]]

Get message history from storage.

Parameters:

Name	Type	Description	Default
`session_id`	`str \| None`	Optional session ID to query history for.	`None`
`limit`	`int \| None`	Maximum number of messages to return.	`None`

Returns:

Type	Description
`list[ChatMessage[Any]]`	List of chat messages from the session.

get_modes `async` ¶

get_modes() -> list[ModeCategory]

Get available mode categories for this agent.

get_resource `async` ¶

get_resource(name: str) -> Any

Get a specific MCP resource by name or URI.

Uses the ExtensionRegistry to query ResourceAccess providers at SESSION scope (POOL + AGENT + SESSION). Falls back to iterating _all_capabilities when the registry is unavailable.

Parameters:

Name	Type	Description	Default
`name`	`str`	Name or URI of the resource to find	required

Raises:

Type	Description
`ToolError`	If resource not found

get_stats `async` ¶

get_stats() -> MessageStats

Get message statistics.

has_pending_injections ¶

has_pending_injections(session_id: str | None = None) -> bool

Check if there are pending injections.

Parameters:

Name	Type	Description	Default
`session_id`	`str \| None`	Optional session ID for SessionPool fallback lookup.	`None`

inject_prompt ¶

inject_prompt(message: str, session_id: str | None = None) -> None

Inject a message into the conversation mid-run.

The message will be injected after the next tool completes (if the agent supports tool hooks). If no tool executes before the run iteration completes, the message is automatically queued for the next iteration.

Deprecated for pooled agents

Use host_context.session_pool.steer() instead.

For standalone agents (no session pool), the injection_manager-based path is used as a fallback.

Parameters:

Name	Type	Description	Default
`message`	`str`	Message to inject	required
`session_id`	`str \| None`	Optional session ID. Falls back to the active run context's session_id if available.	`None`

Example

In a tool implementation:¶

async def my_tool(ctx: AgentContext) -> str: ctx.agent.inject_prompt("Also check the test coverage") return "Changes made"

interrupt `async` ¶

interrupt(run_ctx: AgentRunContext | None = None, session_id: str | None = None) -> None

Interrupt the currently running stream.

Sets the cancelled flag, calls subclass-specific _interrupt(), and emits the interrupted signal.

When pooled, delegates to SessionPool to cancel the active run via RunHandle. When run_ctx is not provided (e.g., from OpenCode abort_session), tries _current_run_ctx_var (ContextVar) first, then falls back to SessionPool's session.current_run_id + get_run() for cross-task access.

Parameters:

Name	Type	Description	Default
`run_ctx`	`AgentRunContext \| None`	Optional per-run context for the stream to interrupt	`None`
`session_id`	`str \| None`	Optional session ID for SessionPool fallback lookup.	`None`

is_busy ¶

is_busy() -> bool

Check if agent is currently processing tasks.

is_cancelled ¶

is_cancelled() -> bool

Check if agent has been cancelled.

Returns:

Type	Description
`bool`	True if cancellation was requested

is_initializing ¶

is_initializing() -> bool

Check if agent is still initializing.

Returns:

Type	Description
`bool`	True if deferred initialization is pending

is_turn_active ¶

is_turn_active() -> bool

Check if a turn is currently running.

Returns:

Type	Description
`bool`	True if there is an active run context, False otherwise.

list_prompts `async` ¶

list_prompts() -> list[Any]

Get all prompts from external capabilities.

Queries the ExtensionRegistry at SESSION scope (falling back to AGENT scope when no session is available) to find FunctionToolsetCapability instances and collects their MCP prompts.

Returns:

Type	Description
`list[Any]`	List of prompt objects from capabilities that provide prompts

list_sessions `async` ¶

list_sessions(*, cwd: str | None = None, limit: int | None = None) -> list[SessionData]

List sessions from storage.

For native agents, queries the pool's session store for all sessions associated with this agent. Fetches conversation titles from storage.

Parameters:

Name	Type	Description	Default
`cwd`	`str \| None`	Filter sessions by working directory (optional). Uses path normalization (resolve) for comparison, so trailing slashes, symlinks, and relative paths are handled.	`None`
`limit`	`int \| None`	Maximum number of sessions to return (optional)	`None`

Returns:

Type	Description
`list[SessionData]`	List of SessionData objects

load_rules `async` ¶

load_rules(project_dir: str | None = None) -> None

Load agent rules from global and project locations.

Searches for AGENTS.md/CLAUDE.md files in: 1. Global: ~/.config/agentpool/AGENTS.md (user-wide rules) 2. Project: {project_dir}/AGENTS.md (project-specific rules)

Both are merged and staged for injection into the first prompt. Uses the agent's execution environment for filesystem access, making this work across local and remote (ACP) environments.

Parameters:

Name	Type	Description	Default
`project_dir`	`str \| None`	Project directory to search for rules. Falls back to env.cwd if not provided.	`None`

load_session `async` ¶

load_session(session_id: str) -> SessionData | None

Load and restore a session from storage.

Loads session data and restores conversation history for this agent. Message history loads independently from session metadata (separate tables).

Parameters:

Name	Type	Description	Default
`session_id`	`str`	Unique identifier for the session to load	required

Returns:

Type	Description
`SessionData \| None`	SessionData if session was found and loaded, None otherwise.

log_message `async` ¶

log_message(message: ChatMessage[Any]) -> None

Handle message from chat signal.

log_session `async` ¶

log_session(
    session_id: str | None = None,
    initial_prompt: str | None = None,
    model: str | None = None,
    parent_session_id: str | None = None,
) -> None

Log conversation to storage if enabled.

Should be called at the start of run_stream() after session_id is set. For native agents, generate session_id first with uuid4(). For wrapped agents (Claude Code), set session_id from SDK session first.

Parameters:

Name	Type	Description	Default
`session_id`	`str \| None`	Optional session ID for the conversation.	`None`
`initial_prompt`	`str \| None`	Optional initial prompt to trigger title generation.	`None`
`model`	`str \| None`	Requested model identifier for this session.	`None`
`parent_session_id`	`str \| None`	Optional parent session ID.	`None`

queue_prompt ¶

queue_prompt(*prompts: PromptCompatible, session_id: str | None = None) -> None

Queue a prompt to be processed after the current run completes.

When called during an active run_stream, the queued prompt will be processed in a continuation loop without exiting the stream. This allows tools or external code to schedule follow-up work.

Deprecated for pooled agents

Use host_context.session_pool.followup() instead.

For standalone agents (no session pool), the injection_manager-based path is used as a fallback.

Parameters:

Name	Type	Description	Default
`*prompts`	`PromptCompatible`	Prompts to queue (same format as run/run_stream)	`()`
`session_id`	`str \| None`	Optional session ID for SessionPool fallback lookup.	`None`

Example

In a tool implementation:¶

async def my_tool(ctx: AgentContext) -> str: ctx.agent.queue_prompt("Now analyze the results") return "Initial work done"

register_worker ¶

register_worker(
    worker: MessageNode[Any, Any],
    *,
    name: str | None = None,
    reset_history_on_run: bool = True,
    pass_message_history: bool = False
) -> Tool

Register another agent as a worker tool.

reset `async` ¶

reset() -> None

Reset agent state (conversation history and tool states).

resume_session `async` ¶

resume_session(session_id: str) -> SessionData | None

Resume a session by ID without loading conversation history.

Unlike load_session, this does NOT populate conversation.chat_messages. It restores the agent's internal state so the conversation can continue, but assumes the client already has the history (or doesn't need it).

This is useful for: - Reconnecting after a disconnect - Automated workflows that don't need UI history - Faster session switching when history display isn't needed

Default implementation calls load_session (subclasses may optimize).

UNSTABLE: This feature is not part of the ACP spec yet.

Parameters:

Name	Type	Description	Default
`session_id`	`str`	Unique identifier for the session to resume	required

Returns:

Type	Description
`SessionData \| None`	SessionData if session was found and resumed, None otherwise

run `async` ¶

run(
    *prompts: PromptCompatible | ChatMessage[Any],
    store_history: bool = True,
    message_id: str | None = None,
    session_id: str | None = None,
    parent_session_id: str | None = None,
    parent_id: str | None = None,
    message_history: MessageHistory | None = None,
    deps: TDeps | None = None,
    input_provider: InputProvider | None = None,
    event_handlers: Sequence[AnyEventHandlerType] | None = None,
    wait_for_connections: bool | None = None,
    depth: int = 0
) -> ChatMessage[TResult]

Run agent with prompt and get response.

This is the standard synchronous run method shared by all agent types. It collects all streaming events from run_stream() and returns the final message.

Parameters:

Name	Type	Description	Default
`prompts`	`PromptCompatible \| ChatMessage[Any]`	User query or instruction	`()`
`store_history`	`bool`	Whether the message exchange should be added to the context window	`True`
`message_id`	`str \| None`	Optional message id for the returned message. Automatically generated if not provided.	`None`
`session_id`	`str \| None`	Optional conversation id for the returned message.	`None`
`parent_session_id`	`str \| None`	Optional parent conversation id.	`None`
`parent_id`	`str \| None`	Parent message id	`None`
`message_history`	`MessageHistory \| None`	Optional MessageHistory object to use instead of agent's own conversation	`None`
`deps`	`TDeps \| None`	Optional dependencies for the agent	`None`
`input_provider`	`InputProvider \| None`	Optional input provider for the agent	`None`
`event_handlers`	`Sequence[AnyEventHandlerType] \| None`	Optional event handlers for this run (overrides agent's handlers)	`None`
`wait_for_connections`	`bool \| None`	Whether to wait for connected agents to complete	`None`
`depth`	`int`	Current delegation depth (0 = top-level run)	`0`

Returns:

Type	Description
`ChatMessage[TResult]`	ChatMessage containing response and run information

Raises:

Type	Description
`RuntimeError`	If no final message received from stream
`UnexpectedModelBehavior`	If the model fails or behaves unexpectedly

run_in_background `async` ¶

run_in_background(
    *prompt: PromptCompatible, max_count: int | None = None, interval: float = 1.0, **kwargs: Any
) -> Task[ChatMessage[TResult] | None]

Run agent continuously in background with prompt or dynamic prompt function.

Parameters:

Name	Type	Description	Default
`prompt`	`PromptCompatible`	Static prompt or function that generates prompts	`()`
`max_count`	`int \| None`	Maximum number of runs (None = infinite)	`None`
`interval`	`float`	Seconds between runs	`1.0`
`**kwargs`	`Any`	Arguments passed to run()	`{}`

run_iter `async` ¶

run_iter(
    *prompt_groups: Sequence[PromptCompatible],
    store_history: bool = True,
    wait_for_connections: bool | None = None
) -> AsyncIterator[ChatMessage[TResult]]

Run agent sequentially on multiple prompt groups.

Parameters:

Name	Type	Description	Default
`prompt_groups`	`Sequence[PromptCompatible]`	Groups of prompts to process sequentially	`()`
`store_history`	`bool`	Whether to store in conversation history	`True`
`wait_for_connections`	`bool \| None`	Whether to wait for connected agents	`None`

Yields:

Type	Description
`AsyncIterator[ChatMessage[TResult]]`	Response messages in sequence

Example

questions = [ ["What is your name?"], ["How old are you?", image1], ["Describe this image", image2], ] async for response in agent.run_iter(*questions): print(response.content)

run_message `async` ¶

run_message(message: ChatMessage[Any], **kwargs: Any) -> ChatMessage[TResult]

Run with an incoming ChatMessage (e.g., from Talk routing).

Extracts content from the message, preserves session_id, and sets parent_id to track the message chain.

Parameters:

Name	Type	Description	Default
`message`	`ChatMessage[Any]`	The incoming ChatMessage to process	required
`**kwargs`	`Any`	Additional arguments passed to run()	`{}`

Returns:

Type	Description
`ChatMessage[TResult]`	Response ChatMessage with message chain tracked via parent_id

run_stream `async` ¶

run_stream(
    *prompts: PromptCompatible,
    store_history: bool = True,
    message_id: str | None = None,
    session_id: str | None = None,
    parent_session_id: str | None = None,
    parent_id: str | None = None,
    message_history: MessageHistory | None = None,
    input_provider: InputProvider | None = None,
    wait_for_connections: bool | None = None,
    deps: TDeps | None = None,
    event_handlers: Sequence[AnyEventHandlerType] | None = None,
    depth: int = 0,
    _run_ctx: AgentRunContext | None = None,
    _skip_pool: bool = False,
    **pydantic_ai_kwargs: Any
) -> AsyncIterator[RichAgentStreamEvent[TResult]]

Run agent with streaming output (the react loop).

This is the self-contained react loop: it creates a per-run context, logs the session, and processes prompts through _run_stream_once(). For native agents, PydanticAI's PendingMessageDrainCapability handles follow-up prompts; for non-native agents, a manual follow-up loop drains the injection queue.

This method can be used standalone (no SessionPool required) or called indirectly via SessionPool-managed agents. Protocol servers that need session lifecycle management should use SessionPool.run_stream() instead.

If prompts are queued via queue_prompt() during execution, they will be processed in sequence without exiting the stream.

Parameters:

Name	Type	Description	Default
`*prompts`	`PromptCompatible`	Input prompts (various formats supported)	`()`
`store_history`	`bool`	Whether to store in history	`True`
`message_id`	`str \| None`	Optional message ID	`None`
`session_id`	`str \| None`	Optional conversation ID	`None`
`parent_session_id`	`str \| None`	Optional parent conversation ID	`None`
`parent_id`	`str \| None`	Optional parent message ID	`None`
`message_history`	`MessageHistory \| None`	Optional message history	`None`
`input_provider`	`InputProvider \| None`	Optional input provider	`None`
`wait_for_connections`	`bool \| None`	Whether to wait for connected agents	`None`
`deps`	`TDeps \| None`	Optional dependencies	`None`
`event_handlers`	`Sequence[AnyEventHandlerType] \| None`	Optional event handlers	`None`
`depth`	`int`	Current delegation depth (0 = top-level run)	`0`
`**pydantic_ai_kwargs`	`Any`	Extra kwargs forwarded to PydanticAI.	`{}`

Yields:

Type	Description
`AsyncIterator[RichAgentStreamEvent[TResult]]`	Stream events during execution

run_stream_with_commands `async` ¶

run_stream_with_commands(
    *prompts: PromptCompatible, **kwargs: Any
) -> AsyncIterator[StreamWithCommandsEvent[TResult]]

Run agent with slash command support.

Separates slash commands from regular prompts, executes commands first, then processes remaining content through the agent.

Parameters:

Name	Type	Description	Default
`*prompts`	`PromptCompatible`	Input prompts (may include slash commands)	`()`
`**kwargs`	`Any`	Additional arguments passed to run_stream	`{}`

Yields:

Type	Description
`AsyncIterator[StreamWithCommandsEvent[TResult]]`	Command events from slash command execution, then stream events from agent

set_mode `async` ¶

set_mode(mode: ModeInfo) -> None

set_mode(mode: str, category_id: ModeCategoryId | str) -> None

set_mode(mode: ModeInfo | str, category_id: ModeCategoryId | str | None = None) -> None

Set a mode within a category.

Parameters:

Name	Type	Description	Default
`mode`	`ModeInfo \| str`	The mode to activate - either a ModeInfo object or mode ID string.	required
`category_id`	`ModeCategoryId \| str \| None`	Category ID. Required if mode is a string, optional if ModeInfo.	`None`

set_model `async` ¶

set_model(model: Model | str) -> None

Set the model for this agent.

set_session_context ¶

set_session_context(session_id: str, parent_session_id: str | None = None) -> None

Set session context for the agent and its event manager.

Parameters:

Name	Type	Description	Default
`session_id`	`str`	The session ID to set	required
`parent_session_id`	`str \| None`	Optional parent session ID	`None`

set_tool_confirmation_mode `async` ¶

set_tool_confirmation_mode(mode: str) -> None

Set tool confirmation mode (agent-specific implementation).

Each agent type handles permission modes differently: - NativeAgent: tool_confirmation_mode ("always", "never", "per_tool") - ACPAgent: auto_approve (bool)

Subclasses should override this method if they support permission modes. The default implementation delegates to _set_mode(mode, "mode").

Parameters:

Name	Type	Description	Default
`mode`	`str`	Mode value in the agent's native format	required

spawn_task ¶

spawn_task(coro_fn: Callable[..., Coroutine[Any, Any, Any]], *args: Any) -> Task[Any]

Schedule a coroutine for background execution on this node.

The created task is tracked in _pending_tasks and automatically removed when it completes. All pending tasks are awaited in __aexit__.

Parameters:

Name	Type	Description	Default
`coro_fn`	`Callable[..., Coroutine[Any, Any, Any]]`	Coroutine function to call.	required
`*args`	`Any`	Positional arguments passed to `coro_fn`.	`()`

Returns:

Type	Description
`Task[Any]`	The created :class:`asyncio.Task`.

stop `async` ¶

stop() -> None

Stop continuous execution if running.

stop_passing_results_to ¶

stop_passing_results_to(other: MessageNode[Any, Any]) -> None

Stop forwarding results to another node.

temporary_state `async` ¶

temporary_state(
    *,
    output_type: type[T] | None = None,
    tools: list[ToolType] | None = None,
    replace_tools: bool = False,
    history: list[AnyPromptType] | SessionQuery | None = None,
    replace_history: bool = False,
    pause_routing: bool = False,
    model: ModelType | None = None
) -> AsyncIterator[Self | Agent[T]]

Temporarily modify agent state.

Parameters:

Name	Type	Description	Default
`output_type`	`type[T] \| None`	Temporary output type to use	`None`
`tools`	`list[ToolType] \| None`	Temporary tools to make available	`None`
`replace_tools`	`bool`	Whether to replace existing tools	`False`
`history`	`list[AnyPromptType] \| SessionQuery \| None`	Conversation history (prompts or query)	`None`
`replace_history`	`bool`	Whether to replace existing history	`False`
`pause_routing`	`bool`	Whether to pause message routing	`False`
`model`	`ModelType \| None`	Temporary model override	`None`

to_structured ¶

to_structured(output_type: type[NewOutputDataT]) -> Agent[TDeps, NewOutputDataT]

Convert this agent to a structured agent.

Warning: This method mutates the agent in place and breaks caching. Changing output type modifies tool definitions sent to the API.

Parameters:

Name	Type	Description	Default
`output_type`	`type[NewOutputDataT]`	Type for structured responses.	required

Returns:

Type	Description
`Agent[TDeps, NewOutputDataT]`	Self (same instance, not a copy)

to_tool ¶

to_tool(
    *,
    name: str | None = None,
    description: str | None = None,
    reset_history_on_run: bool = True,
    pass_message_history: bool = False,
    parent: Agent[Any, Any] | None = None,
    **_kwargs: Any
) -> FunctionTool[OutputDataT]

Create a tool from this agent.

Parameters:

Name	Type	Description	Default
`name`	`str \| None`	Optional tool name override	`None`
`description`	`str \| None`	Optional tool description override	`None`
`reset_history_on_run`	`bool`	Clear agent's history before each run	`True`
`pass_message_history`	`bool`	Pass parent's message history to agent	`False`
`parent`	`Agent[Any, Any] \| None`	Optional parent agent for history/context sharing	`None`

wait `async` ¶

wait() -> ChatMessage[TResult]

Wait for background execution to complete.

AgentKwargs ¶

Bases: TypedDict

Keyword arguments for configuring an Agent instance.

Agent API Reference¶

Agent ¶

agent_pool property writable ¶

agent_type property ¶

command_store property ¶

connection_stats property ¶

display_name property ¶

host_context property ¶

internal_fs property ¶

lifecycle_config property ¶

lifecycle_dimensions property ¶

message_received class-attribute instance-attribute ¶

message_sent class-attribute instance-attribute ¶

model_name property ¶

name property writable ¶

overlay_fs property ¶

session_id property writable ¶

storage property ¶

task_manager property ¶

AgentReset dataclass ¶

InterruptEvent dataclass ¶

__aenter__ async ¶

__aexit__ async ¶

__and__ ¶

__init__ ¶

__rshift__ ¶

connect_to ¶

create_run ¶

create_run_stream async ¶

create_turn ¶

disconnect_all async ¶

emit_agent_event async ¶

ensure_initialized async ¶

fire_and_forget ¶

from_callback classmethod ¶

from_config classmethod ¶

get_active_run_context ¶

get_agentlet async ¶

get_available_models async ¶

get_context ¶

get_mcp_server_info async ¶

get_message_history async ¶

get_modes async ¶

get_resource async ¶

get_stats async ¶

has_pending_injections ¶

inject_prompt ¶

In a tool implementation:¶

interrupt async ¶

is_busy ¶

is_cancelled ¶

is_initializing ¶

is_turn_active ¶

list_prompts async ¶

list_sessions async ¶

load_rules async ¶

load_session async ¶

log_message async ¶

log_session async ¶

queue_prompt ¶

In a tool implementation:¶

register_worker ¶

reset async ¶

resume_session async ¶

run async ¶

run_in_background async ¶

run_iter async ¶

run_message async ¶

run_stream async ¶

run_stream_with_commands async ¶

set_mode async ¶

set_model async ¶

set_session_context ¶

set_tool_confirmation_mode async ¶

spawn_task ¶

stop async ¶

stop_passing_results_to ¶

temporary_state async ¶

to_structured ¶

to_tool ¶

agent_pool `property` `writable` ¶

agent_type `property` ¶

command_store `property` ¶

connection_stats `property` ¶

display_name `property` ¶

host_context `property` ¶

internal_fs `property` ¶

lifecycle_config `property` ¶

lifecycle_dimensions `property` ¶

message_received `class-attribute` `instance-attribute` ¶

message_sent `class-attribute` `instance-attribute` ¶

model_name `property` ¶

name `property` `writable` ¶

overlay_fs `property` ¶

session_id `property` `writable` ¶

storage `property` ¶

task_manager `property` ¶

AgentReset `dataclass` ¶

InterruptEvent `dataclass` ¶

aenter `async` ¶

aexit `async` ¶

and ¶

init ¶

rshift ¶

create_run_stream `async` ¶

disconnect_all `async` ¶

emit_agent_event `async` ¶

ensure_initialized `async` ¶

from_callback `classmethod` ¶

from_config `classmethod` ¶

get_agentlet `async` ¶

get_available_models `async` ¶

get_mcp_server_info `async` ¶

get_message_history `async` ¶

get_modes `async` ¶

get_resource `async` ¶

get_stats `async` ¶

interrupt `async` ¶

list_prompts `async` ¶

list_sessions `async` ¶

load_rules `async` ¶

load_session `async` ¶

log_message `async` ¶

log_session `async` ¶

reset `async` ¶

resume_session `async` ¶

run `async` ¶

run_in_background `async` ¶

run_iter `async` ¶

run_message `async` ¶

run_stream `async` ¶

run_stream_with_commands `async` ¶

set_mode `async` ¶

set_model `async` ¶

set_tool_confirmation_mode `async` ¶

stop `async` ¶

temporary_state `async` ¶

wait `async` ¶