pywry.chat.providers

ACP-conformant provider adapters for PyWry chat.

These classes implement the ACP session lifecycle (initialize, new_session, prompt, cancel) over optional provider SDKs and user-defined callables. They operate on ContentBlock lists from pywry.chat.models and yield SessionUpdate notifications from pywry.chat.updates.

---

Base Provider

pywry.chat.providers.ChatProvider

Bases: ABC

Abstract base class for ACP-conformant chat providers.

Providers adapt third-party LLM clients or external ACP agents to the ACP session lifecycle. They accept content blocks, manage sessions, and yield typed SessionUpdate notifications.

Functions

initialize abstractmethod async

initialize(capabilities: ClientCapabilities) -> AgentCapabilities

Negotiate protocol version and capabilities.

Parameters:

Name	Type	Description	Default
capabilities	ClientCapabilities	Features the client supports.	required

Returns:

Type	Description
AgentCapabilities	Features the agent supports.

new_session abstractmethod async

new_session(cwd: str, mcp_servers: list[dict[str, Any]] | None = None) -> str

Create a new conversation session.

Parameters:

Name	Type	Description	Default
cwd	str	Working directory context for file operations.	required
mcp_servers	list[dict[str, Any]] | None	Optional MCP server configurations.	None

Returns:

Type	Description
str	Unique session identifier.

prompt abstractmethod

prompt(session_id: str, content: list[ContentBlock], cancel_event: Event | None = None) -> AsyncIterator[SessionUpdate]

Process a user prompt and stream session updates.

Parameters:

Name	Type	Description	Default
session_id	str	Active session identifier from new_session().	required
content	list[ContentBlock]	User message content blocks.	required
cancel_event	Event | None	Cooperative cancellation signal.	None

Yields:

Type	Description
SessionUpdate	Typed update notifications.

cancel abstractmethod async

cancel(session_id: str) -> None

Cancel an ongoing prompt turn.

Parameters:

Name	Type	Description	Default
session_id	str	Session to cancel.	required

load_session async

load_session(session_id: str, cwd: str) -> str

Resume a prior session.

Parameters:

Name	Type	Description	Default
session_id	str	Session identifier to restore.	required
cwd	str	Working directory context.	required

Returns:

Type	Description
str	The restored session identifier.

Raises:

Type	Description
NotImplementedError	If the provider does not support session loading.

set_config_option async

set_config_option(session_id: str, option_id: str, value: str) -> list[SessionConfigOption]

Change a config option, return full config state.

Parameters:

Name	Type	Description	Default
session_id	str	Active session identifier.	required
option_id	str	Config option to change.	required
value	str	New value.	required

Returns:

Type	Description
list[SessionConfigOption]	Complete set of config options with current values.

set_mode async

set_mode(session_id: str, mode_id: str) -> None

Switch agent mode.

Parameters:

Name	Type	Description	Default
session_id	str	Active session identifier.	required
mode_id	str	Mode to activate.	required

---

Provider Implementations

pywry.chat.providers.openai.OpenAIProvider

OpenAIProvider(**kwargs: Any)

Bases: ChatProvider

Provider backed by the openai async client.

Parameters:

Name	Type	Description	Default
**kwargs	Any	Keyword arguments forwarded to openai.AsyncOpenAI (e.g. api_key, base_url).	{}

Raises:

Type	Description
ImportError	If the openai package is not installed.

Attributes

_client instance-attribute

_client = AsyncOpenAI(**kwargs)

_sessions instance-attribute

_sessions: dict[str, dict[str, Any]] = {}

Functions

initialize async

initialize(capabilities: ClientCapabilities) -> AgentCapabilities

Return text-only prompt capabilities.

Parameters:

Name	Type	Description	Default
capabilities	ClientCapabilities	Client features (unused by OpenAI adapter).	required

Returns:

Type	Description
AgentCapabilities	Advertises text and image prompt support.

new_session async

new_session(cwd: str, mcp_servers: list[dict[str, Any]] | None = None) -> str

Create an in-memory session tracker.

Parameters:

Name	Type	Description	Default
cwd	str	Working directory context.	required
mcp_servers	list[dict[str, Any]] | None	MCP server configs (unused by OpenAI adapter).	None

Returns:

Type	Description
str	Unique session identifier.

prompt async

prompt(session_id: str, content: list[ContentBlock], cancel_event: Event | None = None) -> AsyncIterator[SessionUpdate]

Stream response chunks as AgentMessageUpdate.

Parameters:

Name	Type	Description	Default
session_id	str	Active session identifier.	required
content	list[ContentBlock]	User message content blocks.	required
cancel_event	Event | None	Cooperative cancellation signal.	None

Yields:

Type	Description
SessionUpdate	AgentMessageUpdate for each text chunk.

cancel async

cancel(session_id: str) -> None

Cancel is handled cooperatively via cancel_event.

Parameters:

Name	Type	Description	Default
session_id	str	Session to cancel.	required

load_session async

load_session(session_id: str, cwd: str) -> str

Resume a prior session.

Parameters:

Name	Type	Description	Default
session_id	str	Session identifier to restore.	required
cwd	str	Working directory context.	required

Returns:

Type	Description
str	The restored session identifier.

Raises:

Type	Description
NotImplementedError	If the provider does not support session loading.

set_config_option async

set_config_option(session_id: str, option_id: str, value: str) -> list[SessionConfigOption]

Change a config option, return full config state.

Parameters:

Name	Type	Description	Default
session_id	str	Active session identifier.	required
option_id	str	Config option to change.	required
value	str	New value.	required

Returns:

Type	Description
list[SessionConfigOption]	Complete set of config options with current values.

set_mode async

set_mode(session_id: str, mode_id: str) -> None

Switch agent mode.

Parameters:

Name	Type	Description	Default
session_id	str	Active session identifier.	required
mode_id	str	Mode to activate.	required

pywry.chat.providers.anthropic.AnthropicProvider

AnthropicProvider(**kwargs: Any)

Bases: ChatProvider

Provider backed by the anthropic async client.

Parameters:

Name	Type	Description	Default
**kwargs	Any	Keyword arguments forwarded to anthropic.AsyncAnthropic (e.g. api_key).	{}

Raises:

Type	Description
ImportError	If the anthropic package is not installed.

Attributes

_client instance-attribute

_client = AsyncAnthropic(**kwargs)

_sessions instance-attribute

_sessions: dict[str, dict[str, Any]] = {}

Functions

initialize async

initialize(capabilities: ClientCapabilities) -> AgentCapabilities

Return text+image prompt capabilities.

Parameters:

Name	Type	Description	Default
capabilities	ClientCapabilities	Client features (unused by Anthropic adapter).	required

Returns:

Type	Description
AgentCapabilities	Advertises text and image prompt support.

new_session async

new_session(cwd: str, mcp_servers: list[dict[str, Any]] | None = None) -> str

Create an in-memory session tracker.

Parameters:

Name	Type	Description	Default
cwd	str	Working directory context.	required
mcp_servers	list[dict[str, Any]] | None	MCP server configs (unused by Anthropic adapter).	None

Returns:

Type	Description
str	Unique session identifier.

prompt async

prompt(session_id: str, content: list[ContentBlock], cancel_event: Event | None = None) -> AsyncIterator[SessionUpdate]

Stream response chunks as AgentMessageUpdate.

Parameters:

Name	Type	Description	Default
session_id	str	Active session identifier.	required
content	list[ContentBlock]	User message content blocks.	required
cancel_event	Event | None	Cooperative cancellation signal.	None

Yields:

Type	Description
SessionUpdate	AgentMessageUpdate for each text chunk.

cancel async

cancel(session_id: str) -> None

Cancel is handled cooperatively via cancel_event.

Parameters:

Name	Type	Description	Default
session_id	str	Session to cancel.	required

load_session async

load_session(session_id: str, cwd: str) -> str

Resume a prior session.

Parameters:

Name	Type	Description	Default
session_id	str	Session identifier to restore.	required
cwd	str	Working directory context.	required

Returns:

Type	Description
str	The restored session identifier.

Raises:

Type	Description
NotImplementedError	If the provider does not support session loading.

set_config_option async

set_config_option(session_id: str, option_id: str, value: str) -> list[SessionConfigOption]

Change a config option, return full config state.

Parameters:

Name	Type	Description	Default
session_id	str	Active session identifier.	required
option_id	str	Config option to change.	required
value	str	New value.	required

Returns:

Type	Description
list[SessionConfigOption]	Complete set of config options with current values.

set_mode async

set_mode(session_id: str, mode_id: str) -> None

Switch agent mode.

Parameters:

Name	Type	Description	Default
session_id	str	Active session identifier.	required
mode_id	str	Mode to activate.	required

pywry.chat.providers.callback.CallbackProvider

CallbackProvider(prompt_fn: Any = None)

Bases: ChatProvider

Provider backed by user-supplied Python callables.

Parameters:

Name	Type	Description	Default
prompt_fn	callable	Callable invoked on each prompt. It may be sync or async, and should yield SessionUpdate objects or plain strings (which are wrapped as AgentMessageUpdate).	None

Attributes

_prompt_fn instance-attribute

_prompt_fn = prompt_fn

Functions

initialize async

initialize(capabilities: ClientCapabilities) -> AgentCapabilities

Return minimal capabilities.

Parameters:

Name	Type	Description	Default
capabilities	ClientCapabilities	Client features (unused).	required

Returns:

Type	Description
AgentCapabilities	Default capabilities.

new_session async

new_session(cwd: str, mcp_servers: list[dict[str, Any]] | None = None) -> str

Create a lightweight session identifier.

Parameters:

Name	Type	Description	Default
cwd	str	Working directory context.	required
mcp_servers	list[dict[str, Any]] | None	MCP server configs (unused).	None

Returns:

Type	Description
str	Unique session identifier.

prompt async

prompt(session_id: str, content: list[ContentBlock], cancel_event: Event | None = None) -> AsyncIterator[SessionUpdate]

Invoke the user callback and yield session updates.

Plain strings yielded by the callback are wrapped as AgentMessageUpdate. SessionUpdate instances are yielded as-is.

Parameters:

Name	Type	Description	Default
session_id	str	Active session identifier.	required
content	list[ContentBlock]	User message content blocks.	required
cancel_event	Event | None	Cooperative cancellation signal.	None

Yields:

Type	Description
SessionUpdate	Updates from the callback.

_iter_result async staticmethod

_iter_result(result: Any, cancel_event: Any) -> AsyncIterator[SessionUpdate]

Iterate a sync or async result, wrapping strings.

cancel async

cancel(session_id: str) -> None

Cancel is handled cooperatively via cancel_event.

Parameters:

Name	Type	Description	Default
session_id	str	Session to cancel.	required

load_session async

load_session(session_id: str, cwd: str) -> str

Resume a prior session.

Parameters:

Name	Type	Description	Default
session_id	str	Session identifier to restore.	required
cwd	str	Working directory context.	required

Returns:

Type	Description
str	The restored session identifier.

Raises:

Type	Description
NotImplementedError	If the provider does not support session loading.

set_config_option async

set_config_option(session_id: str, option_id: str, value: str) -> list[SessionConfigOption]

Change a config option, return full config state.

Parameters:

Name	Type	Description	Default
session_id	str	Active session identifier.	required
option_id	str	Config option to change.	required
value	str	New value.	required

Returns:

Type	Description
list[SessionConfigOption]	Complete set of config options with current values.

set_mode async

set_mode(session_id: str, mode_id: str) -> None

Switch agent mode.

Parameters:

Name	Type	Description	Default
session_id	str	Active session identifier.	required
mode_id	str	Mode to activate.	required

pywry.chat.providers.magentic.MagenticProvider

MagenticProvider(model: Any, **kwargs: Any)

Bases: ChatProvider

Provider wrapping a magentic ChatModel backend.

Parameters:

Name	Type	Description	Default
model	ChatModel | str	A pre-configured magentic ChatModel instance, or a model name string (creates an OpenaiChatModel).	required
**kwargs	Any	Extra keyword arguments forwarded to OpenaiChatModel when model is a string.	{}

Raises:

Type	Description
ImportError	If the magentic package is not installed.

Attributes

_model instance-attribute

_model = model

_sessions instance-attribute

_sessions: dict[str, dict[str, Any]] = {}

Functions

initialize async

initialize(capabilities: ClientCapabilities) -> AgentCapabilities

Return text-only prompt capabilities.

Parameters:

Name	Type	Description	Default
capabilities	ClientCapabilities	Client features (unused).	required

Returns:

Type	Description
AgentCapabilities	Default capabilities.

new_session async

new_session(cwd: str, mcp_servers: list[dict[str, Any]] | None = None) -> str

Create an in-memory session tracker.

Parameters:

Name	Type	Description	Default
cwd	str	Working directory context.	required
mcp_servers	list[dict[str, Any]] | None	MCP server configs (unused).	None

Returns:

Type	Description
str	Unique session identifier.

prompt async

prompt(session_id: str, content: list[ContentBlock], cancel_event: Event | None = None) -> AsyncIterator[SessionUpdate]

Stream response chunks as AgentMessageUpdate.

Parameters:

Name	Type	Description	Default
session_id	str	Active session identifier.	required
content	list[ContentBlock]	User message content blocks.	required
cancel_event	Event | None	Cooperative cancellation signal.	None

Yields:

Type	Description
SessionUpdate	AgentMessageUpdate for each text chunk.

cancel async

cancel(session_id: str) -> None

Cancel is handled cooperatively via cancel_event.

Parameters:

Name	Type	Description	Default
session_id	str	Session to cancel.	required

load_session async

load_session(session_id: str, cwd: str) -> str

Resume a prior session.

Parameters:

Name	Type	Description	Default
session_id	str	Session identifier to restore.	required
cwd	str	Working directory context.	required

Returns:

Type	Description
str	The restored session identifier.

Raises:

Type	Description
NotImplementedError	If the provider does not support session loading.

set_config_option async

set_config_option(session_id: str, option_id: str, value: str) -> list[SessionConfigOption]

Change a config option, return full config state.

Parameters:

Name	Type	Description	Default
session_id	str	Active session identifier.	required
option_id	str	Config option to change.	required
value	str	New value.	required

Returns:

Type	Description
list[SessionConfigOption]	Complete set of config options with current values.

set_mode async

set_mode(session_id: str, mode_id: str) -> None

Switch agent mode.

Parameters:

Name	Type	Description	Default
session_id	str	Active session identifier.	required
mode_id	str	Mode to activate.	required

pywry.chat.providers.stdio.StdioProvider

StdioProvider(command: str, args: list[str] | None = None, env: dict[str, str] | None = None)

Bases: ChatProvider

Connect to an external ACP agent via stdio JSON-RPC 2.0.

Parameters:

Name	Type	Description	Default
command	str	Executable to spawn (e.g. "claude").	required
args	list[str] | None	Command-line arguments.	None
env	dict[str, str] | None	Environment variable overrides.	None

Examples:

>>> provider = StdioProvider(command="claude", args=["--agent"])
>>> caps = await provider.initialize(ClientCapabilities())
>>> session_id = await provider.new_session("/path/to/project")

Attributes

_command instance-attribute

_command = command

_args instance-attribute

_args = args or []

_env instance-attribute

_env = env

_process instance-attribute

_process: Process | None = None

_pending instance-attribute

_pending: dict[str | int, Future[Any]] = {}

_reader_task instance-attribute

_reader_task: Task[None] | None = None

_stderr_task instance-attribute

_stderr_task: Task[None] | None = None

_update_queues instance-attribute

_update_queues: dict[str, Queue[Any]] = {}

Functions

_ensure_started async

_ensure_started() -> Process

Spawn the subprocess if not already running.

Returns:

Type	Description
Process	The running subprocess.

_drain_stderr async

_drain_stderr() -> None

Continuously consume stderr so the subprocess can't block.

_fail_pending_requests

_fail_pending_requests(exc: BaseException) -> None

Fail every pending JSON-RPC request future with exc.

Called when the read loop terminates (subprocess exited, stdout closed, reader cancelled, etc.) so callers blocked on _send_request wake up with a real error instead of hanging forever.

_dispatch_rpc_message async

_dispatch_rpc_message(msg: dict[str, Any]) -> None

Route one decoded JSON-RPC message to the right handler.

_read_loop async

_read_loop() -> None

Read JSON-RPC messages from stdout line-by-line.

_send_request async

_send_request(method: str, params: dict[str, Any] | None = None) -> Any

Send a JSON-RPC request and wait for the response.

Parameters:

Name	Type	Description	Default
method	str	RPC method name.	required
params	dict[str, Any] | None	Method parameters.	None

Returns:

Type	Description
Any	The result from the response.

_send_notification async

_send_notification(method: str, params: dict[str, Any] | None = None) -> None

Send a JSON-RPC notification (no response expected).

Parameters:

Name	Type	Description	Default
method	str	RPC method name.	required
params	dict[str, Any] | None	Method parameters.	None

_handle_notification async

_handle_notification(msg: dict[str, Any]) -> None

Route agent notifications to the appropriate session queue.

Parameters:

Name	Type	Description	Default
msg	dict[str, Any]	JSON-RPC notification message.	required

_handle_agent_request async

_handle_agent_request(msg: dict[str, Any]) -> None

Handle requests from the agent (permission, fs, terminal).

Parameters:

Name	Type	Description	Default
msg	dict[str, Any]	JSON-RPC request message from the agent.	required

initialize async

initialize(capabilities: Any) -> Any

Send initialize to the agent subprocess.

Parameters:

Name	Type	Description	Default
capabilities	ClientCapabilities	Client features to advertise.	required

Returns:

Type	Description
AgentCapabilities	Features the agent supports.

new_session async

new_session(cwd: str, mcp_servers: list[dict[str, Any]] | None = None) -> str

Send session/new to the agent.

Parameters:

Name	Type	Description	Default
cwd	str	Working directory context.	required
mcp_servers	list[dict[str, Any]] | None	MCP server configurations.	None

Returns:

Type	Description
str	Session identifier from the agent.

_serialize_content_blocks staticmethod

_serialize_content_blocks(content: list[Any]) -> list[dict[str, Any]]

Serialize content blocks (Pydantic models or raw dicts).

_update_type_map staticmethod

_update_type_map() -> dict[str, type]

Return the discriminator → model-class map for stdio updates.

_settle_prompt_future async

_settle_prompt_future(prompt_future: Future[Any]) -> None

Settle the in-flight session/prompt request.

Always called from prompt()'s finally block so exceptions from the agent surface to the caller and _pending can't leak. On cancellation the future is cancelled first; the _pending entry is either cleared by the read loop (when the agent acks) or failed en masse by _fail_pending_requests when the subprocess exits.

prompt async

prompt(session_id: str, content: list[Any], cancel_event: Event | None = None) -> Any

Send session/prompt and yield session updates.

Parameters:

Name	Type	Description	Default
session_id	str	Active session identifier.	required
content	list[ContentBlock]	User message content blocks.	required
cancel_event	Event | None	Cooperative cancellation signal.	None

Yields:

Type	Description
SessionUpdate	Typed update notifications from the agent.

_parse_update staticmethod

_parse_update(update: dict[str, Any], update_map: Mapping[str, type]) -> Any

Build a SessionUpdate model from a raw update dict.

cancel async

cancel(session_id: str) -> None

Send session/cancel notification.

Parameters:

Name	Type	Description	Default
session_id	str	Session to cancel.	required

close async

close() -> None

Terminate the subprocess and clean up.

Call this when done with the provider to release system resources.

load_session async

load_session(session_id: str, cwd: str) -> str

Resume a prior session.

Parameters:

Name	Type	Description	Default
session_id	str	Session identifier to restore.	required
cwd	str	Working directory context.	required

Returns:

Type	Description
str	The restored session identifier.

Raises:

Type	Description
NotImplementedError	If the provider does not support session loading.

set_config_option async

set_config_option(session_id: str, option_id: str, value: str) -> list[SessionConfigOption]

Change a config option, return full config state.

Parameters:

Name	Type	Description	Default
session_id	str	Active session identifier.	required
option_id	str	Config option to change.	required
value	str	New value.	required

Returns:

Type	Description
list[SessionConfigOption]	Complete set of config options with current values.

set_mode async

set_mode(session_id: str, mode_id: str) -> None

Switch agent mode.

Parameters:

Name	Type	Description	Default
session_id	str	Active session identifier.	required
mode_id	str	Mode to activate.	required

pywry.chat.providers.deepagent.DeepagentProvider

DeepagentProvider(agent: Any = None, *, model: str = 'anthropic:claude-sonnet-4-6', tools: list[Any] | None = None, mcp_servers: dict[str, dict[str, Any]] | None = None, system_prompt: str = '', replace_system_prompt: bool = False, checkpointer: Any = None, store: Any = None, memory: list[str] | None = None, interrupt_on: dict[str, Any] | None = None, backend: Any = None, subagents: list[dict[str, Any]] | None = None, skills: list[str] | None = None, middleware: list[Any] | None = None, auto_checkpointer: bool = True, auto_store: bool = True, recursion_limit: int = 50, **kwargs: Any)

Bases: ChatProvider

Provider wrapping a LangChain Deep Agents CompiledGraph.

Parameters:

Name	Type	Description	Default
agent	CompiledGraph or None	A pre-built agent from create_deep_agent(). If None, the provider calls create_deep_agent() internally using the other parameters.	None
model	str	Model identifier in provider:model format.	'anthropic:claude-sonnet-4-6'
tools	list[callable] or None	Local LangChain-compatible tool callables. Merged with any MCP-served tools before the agent is built.	None
mcp_servers	dict[str, dict] or None	MCP servers the agent should connect to, in the langchain_mcp_adapters.client.MultiServerMCPClient config format — one entry per server, keyed by a short name. Example:: { "pywry": { "transport": "streamable_http", "url": "http://127.0.0.1:8765/mcp", }, "fs": { "transport": "stdio", "command": "uvx", "args": ["mcp-server-filesystem", "/tmp"], }, } On first agent build the provider connects to every server and converts the exposed MCP tools into LangChain tools, merging them with tools before calling create_deep_agent. Requires the pywry[deepagent] extra (which pulls in langchain-mcp-adapters).	None
system_prompt	str	System instructions for the agent. By default this is appended to PYWRY_SYSTEM_PROMPT (the general-purpose guidance about the PyWry chat environment). Pass replace_system_prompt=True to fully override instead — useful when the caller's agent has a narrow tool surface and needs tighter output constraints than the general prompt allows.	''
replace_system_prompt	bool	If True, system_prompt replaces PYWRY_SYSTEM_PROMPT entirely instead of being appended. Defaults to False.	False
checkpointer	Any or None	LangGraph checkpointer for session persistence. If None and auto_checkpointer=True, one is created based on PyWry's state backend.	None
store	Any or None	LangGraph Memory Store for cross-session knowledge persistence. If None and auto_store=True, an InMemoryStore is created so the agent retains knowledge within the process lifetime.	None
memory	list[str] or None	Paths to memory files (e.g. ["/AGENTS.md"]) that the agent can read and write for persistent context.	None
interrupt_on	dict or None	Tool names that require human approval before execution.	None
backend	Any or None	Deep Agents filesystem backend.	None
subagents	list[dict] or None	Subagent configurations.	None
skills	list[str] or None	File paths to Deep Agents skill markdown files that the agent can reference on demand. PyWry ships seventeen of these under pywry.mcp.skills (tvchart, chat_agent, events, component_reference, authentication, etc.) — build the list with pathlib.Path(pywry.mcp.skills.__file__).parent / "<skill>" / "SKILL.md". Forwarded verbatim to create_deep_agent(skills=...).	None
middleware	list or None	Deep Agents middleware callables.	None
auto_checkpointer	bool	Auto-select checkpointer based on PyWry state backend. Runs on first _build_agent() so callers that bypass the async initialize() still get conversation-history persistence.	True
auto_store	bool	Auto-create an InMemoryStore if no store is provided. The store enables cross-thread memory persistence within the process lifetime.	True
recursion_limit	int	LangGraph recursion limit per prompt turn. Every tool call costs 2-3 graph steps, so the default (50) leaves headroom for multi-tool turns without hiding pathological loops. LangGraph's own default is 25.	50

Attributes

_agent instance-attribute

_agent = agent

_model instance-attribute

_model = model

_tools instance-attribute

_tools = tools or []

_mcp_servers instance-attribute

_mcp_servers = mcp_servers or {}

_mcp_tools instance-attribute

_mcp_tools: list[Any] = []

_system_prompt instance-attribute

_system_prompt = system_prompt

_replace_system_prompt instance-attribute

_replace_system_prompt = replace_system_prompt

_checkpointer instance-attribute

_checkpointer = checkpointer

_store instance-attribute

_store = store

_memory instance-attribute

_memory = memory

_interrupt_on instance-attribute

_interrupt_on = interrupt_on

_backend instance-attribute

_backend = backend

_subagents instance-attribute

_subagents = subagents

_skills instance-attribute

_skills = skills

_middleware instance-attribute

_middleware = middleware

_auto_checkpointer instance-attribute

_auto_checkpointer = auto_checkpointer

_auto_store instance-attribute

_auto_store = auto_store

_recursion_limit instance-attribute

_recursion_limit = recursion_limit

_kwargs instance-attribute

_kwargs = kwargs

_sessions instance-attribute

_sessions: dict[str, str] = {}

Functions

initialize async

initialize(capabilities: ClientCapabilities) -> AgentCapabilities

Build the agent and configure the checkpointer.

Parameters:

Name	Type	Description	Default
capabilities	ClientCapabilities	Client features.	required

Returns:

Type	Description
AgentCapabilities	Agent features.

_create_checkpointer

_create_checkpointer() -> Any

_create_store

_create_store() -> Any

_load_mcp_tools

_load_mcp_tools() -> list[Any]

Connect to configured MCP servers and load their tools.

Uses langchain_mcp_adapters.client.MultiServerMCPClient to connect to every server in self._mcp_servers and convert the exposed MCP tools into LangChain tools. Returns an empty list when no servers are configured or the bridge package is missing.

_build_agent_kwargs

_build_agent_kwargs(merged_tools: list[Any], system_prompt: str) -> dict[str, Any]

Assemble the kwargs dict for create_deep_agent from provider state.

_build_agent

_build_agent() -> Any

new_session async

new_session(cwd: str, mcp_servers: list[dict[str, Any]] | None = None) -> str

Create a session mapped to a LangGraph thread_id.

Parameters:

Name	Type	Description	Default
cwd	str	Working directory context.	required
mcp_servers	list[dict] or None	ACP-style MCP server descriptors. Each entry is converted to the MultiServerMCPClient config format and merged into the provider's existing mcp_servers map; the next agent build picks them up.	None

Returns:

Type	Description
str	Session identifier.

load_session async

load_session(session_id: str, cwd: str) -> str

Resume a session using its LangGraph thread_id.

Parameters:

Name	Type	Description	Default
session_id	str	Session to restore.	required
cwd	str	Working directory.	required

Returns:

Type	Description
str	The restored session identifier.

prompt async

prompt(session_id: str, content: list[ContentBlock], cancel_event: Event | None = None) -> AsyncIterator[SessionUpdate]

Stream LangGraph events as ACP SessionUpdate objects.

Parameters:

Name	Type	Description	Default
session_id	str	Active session identifier.	required
content	list[ContentBlock]	User message content blocks.	required
cancel_event	Event or None	Cooperative cancellation signal.	None

Yields:

Type	Description
SessionUpdate	Typed update notifications.

_stream_chat_model async

_stream_chat_model(event: dict[str, Any], text_filter: _ToolCallTextFilter | None) -> AsyncIterator[SessionUpdate]

Yield ThinkingUpdate / AgentMessageUpdate for on_chat_model_stream.

_stream_tool_start async

_stream_tool_start(event: dict[str, Any]) -> AsyncIterator[SessionUpdate]

Yield StatusUpdate (write_todos) or ToolCallUpdate for on_tool_start.

_stream_misc_event async

_stream_misc_event(event: dict[str, Any], kind: str) -> AsyncIterator[SessionUpdate]

Yield updates for the smaller event kinds (errors, status, subagent).

_dispatch_stream_event async

_dispatch_stream_event(event: dict[str, Any], text_filter: _ToolCallTextFilter | None = None) -> AsyncIterator[SessionUpdate]

Route a single LangGraph streaming event to the matching update.

text_filter is the per-prompt stateful stripper that removes leaked functions.<name>:N{...} tool-call markup from the assistant text stream. Optional so tests / direct callers can skip it; production paths in prompt() always pass one.

_handle_tool_end async

_handle_tool_end(event: dict[str, Any]) -> AsyncIterator[SessionUpdate]

Handle on_tool_end events, including write_todos → PlanUpdate.

cancel async

cancel(session_id: str) -> None

Cancel is handled cooperatively via cancel_event.

Parameters:

Name	Type	Description	Default
session_id	str	Session to cancel.	required

truncate_session

truncate_session(session_id: str, kept_messages: list[Any]) -> None

Discard the LangGraph checkpointer state for a session.

Called by ChatManager when the user edits or resends a message in the middle of a thread. The next prompt call will rebuild the agent state from the surviving messages — but LangGraph's checkpointer keeps appending, so the cleanest fix is to forget the prior state entirely for this thread.

Parameters:

Name	Type	Description	Default
session_id	str	ChatManager session id (also used as the LangGraph thread_id).	required
kept_messages	list	Messages that survive in the UI; passed for callers that may want to seed an alternate store. This implementation only uses the session_id.	required

set_config_option async

set_config_option(session_id: str, option_id: str, value: str) -> list[SessionConfigOption]

Change a config option, return full config state.

Parameters:

Name	Type	Description	Default
session_id	str	Active session identifier.	required
option_id	str	Config option to change.	required
value	str	New value.	required

Returns:

Type	Description
list[SessionConfigOption]	Complete set of config options with current values.

set_mode async

set_mode(session_id: str, mode_id: str) -> None

Switch agent mode.

Parameters:

Name	Type	Description	Default
session_id	str	Active session identifier.	required
mode_id	str	Mode to activate.	required

---

Factory

pywry.chat.providers.get_provider

get_provider(name: str, **kwargs: Any) -> ChatProvider

Create a provider instance by name.

Parameters:

Name	Type	Description	Default
name	str	Provider name. Supported values: "openai", "anthropic", "callback", "magentic", "stdio".	required
**kwargs	Any	Passed to the provider constructor.	{}

Returns:

Type	Description
ChatProvider	Instantiated provider.

Raises:

Type	Description
ValueError	If provider name is unknown.