pydantic_acp API

This page documents the public surface re-exported by pydantic_acp.

Functions

create_acp_agent(agent=None, *, agent_factory=None, agent_source=None, config=None, projection_maps=None)

Source code in packages/adapters/pydantic-acp/src/pydantic_acp/runtime/server.py

def create_acp_agent(
    agent: PydanticAgent[AgentDepsT, OutputDataT] | None = None,
    *,
    agent_factory: AgentFactory[AgentDepsT, OutputDataT] | None = None,
    agent_source: AgentSource[AgentDepsT, OutputDataT] | None = None,
    config: AdapterConfig | None = None,
    projection_maps: Sequence[ProjectionMap | HookProjectionMap] | None = None,
) -> PydanticAcpAgent[AgentDepsT, OutputDataT]:
    resolved_source = _resolve_agent_source(
        agent=agent,
        agent_factory=agent_factory,
        agent_source=agent_source,
    )
    resolved_config = _resolve_config(
        config=config,
        agent_name=agent.name if agent is not None else None,
        projection_maps=projection_maps,
    )
    adapter = PydanticAcpAgent(resolved_source, config=resolved_config)
    return adapter

create_acp_model(*, acp_agent=None, acp_command=None, model_name=None, cwd=None, env=None, stderr_mode='inherit', terminate_timeout=5.0, prompt_renderer=None, history_mode='latest_user', delegate_client=None, enable_pydantic_acp_meta=None, settings=None, profile=None)

Create a Pydantic AI model backed by an ACP agent or ACP stdio command.

Exactly one of acp_agent or acp_command must be provided. Passing model_name=None leaves ACP model selection to the remote agent's session default and does not send a session/set_config_option request for "model".

Source code in packages/adapters/pydantic-acp/src/pydantic_acp/factory.py

def create_acp_model(
    *,
    acp_agent: AcpAgent | None = None,
    acp_command: Sequence[str] | None = None,
    model_name: str | None = None,
    cwd: str | Path | None = None,
    env: Mapping[str, str] | None = None,
    stderr_mode: CommandStderrMode = "inherit",
    terminate_timeout: float = 5.0,
    prompt_renderer: AcpPromptRenderer | None = None,
    history_mode: HistoryMode = "latest_user",
    delegate_client: AcpClient | None = None,
    enable_pydantic_acp_meta: bool | None = None,
    settings: ModelSettings | None = None,
    profile: ModelProfileSpec | None = None,
) -> AcpModel:
    """Create a Pydantic AI model backed by an ACP agent or ACP stdio command.

    Exactly one of ``acp_agent`` or ``acp_command`` must be provided. Passing
    ``model_name=None`` leaves ACP model selection to the remote agent's session
    default and does not send a ``session/set_config_option`` request for ``"model"``.
    """

    command = _normalize_command(acp_command)
    resolved_cwd = Path.cwd() if cwd is None else Path(cwd)
    if command is None:
        if acp_agent is None:
            raise ValueError("Exactly one of acp_agent or acp_command must be provided.")
        source_agent = acp_agent
    else:
        if acp_agent is not None:
            raise ValueError("Exactly one of acp_agent or acp_command must be provided.")
        source_agent = AcpCommandAgent(
            options=AcpCommandOptions(
                command=command,
                cwd=resolved_cwd,
                env=env,
                stderr_mode=stderr_mode,
                terminate_timeout=terminate_timeout,
            ),
        )

    provider = AcpProvider(
        acp_agent=source_agent,
        host_client=delegate_client,
        cwd=resolved_cwd,
        prompt_renderer=prompt_renderer,
        history_mode=history_mode,
        enable_pydantic_acp_meta=enable_pydantic_acp_meta,
    )
    return provider.model(
        model_name,
        settings=settings,
        profile=profile,
        history_mode=history_mode,
    )

run_acp(agent=None, *, agent_factory=None, agent_source=None, config=None, projection_maps=None)

Source code in packages/adapters/pydantic-acp/src/pydantic_acp/runtime/server.py

def run_acp(
    agent: PydanticAgent[AgentDepsT, OutputDataT] | None = None,
    *,
    agent_factory: AgentFactory[AgentDepsT, OutputDataT] | None = None,
    agent_source: AgentSource[AgentDepsT, OutputDataT] | None = None,
    config: AdapterConfig | None = None,
    projection_maps: Sequence[ProjectionMap | HookProjectionMap] | None = None,
) -> None:
    adapter = create_acp_agent(
        agent=agent,
        agent_factory=agent_factory,
        agent_source=agent_source,
        config=config,
        projection_maps=projection_maps,
    )
    asyncio.run(run_agent(adapter))

compose_projection_maps(projection_maps)

Source code in packages/adapters/pydantic-acp/src/pydantic_acp/projection.py

def compose_projection_maps(
    projection_maps: Sequence[ProjectionMap] | None,
) -> ProjectionMap | None:
    if projection_maps is None:
        return None
    if len(projection_maps) == 0:
        return None
    if len(projection_maps) == 1:
        return projection_maps[0]
    return CompositeProjectionMap(maps=tuple(projection_maps))

Core Classes And Data Types

AdapterConfig(*, agent_name=DEFAULT_AGENT_NAME, agent_title=DEFAULT_AGENT_TITLE, agent_version=DEFAULT_AGENT_VERSION, allow_model_selection=False, approval_bridge=NativeApprovalBridge(), approval_state_provider=None, capability_bridges=list(), config_options_provider=None, enable_generic_tool_projection=True, enable_model_config_option=True, host_access_policy=None, hook_projection_map=HookProjectionMap(), models_provider=None, modes_provider=None, native_plan_additional_instructions=None, native_plan_persistence_provider=None, plan_id='acpkit.plan', plan_provider=None, plan_update_mode='full', prompt_capabilities=AdapterPromptCapabilities(), prompt_model_override_provider=None, replay_history_on_load=True, slash_command_provider=None, available_models=list(), session_store=MemorySessionStore(), output_serializer=DefaultOutputSerializer(), projection_maps=tuple(), tool_classifier=DefaultToolClassifier()) dataclass

AdapterModel(*, model_id, name, override, description=None) dataclass

AdapterPromptCapabilities(*, audio=True, image=True, embedded_context=True) dataclass

AcpSessionContext(*, session_id, cwd, created_at, updated_at, additional_directories=(), title=None, session_model_id=None, message_history_json=None, plan_markdown=None, plan_entries=list(), active_plan_id=None, config_values=dict(), mcp_servers=list(), metadata=dict(), transcript=list(), client=None, client_capabilities=None) dataclass

supports_config_options()

Return whether the connected client accepts session config options.

Source code in packages/adapters/pydantic-acp/src/pydantic_acp/session/state.py

def supports_config_options(self) -> bool:
    """Return whether the connected client accepts session config options."""
    capabilities = self.client_capabilities
    if capabilities is None:
        return True
    return capabilities.session is not None and capabilities.session.config_options is not None

supports_boolean_config_options()

Return whether the connected client accepts boolean config options.

Source code in packages/adapters/pydantic-acp/src/pydantic_acp/session/state.py

def supports_boolean_config_options(self) -> bool:
    """Return whether the connected client accepts boolean config options."""
    capabilities = self.client_capabilities
    if capabilities is None:
        return True
    session_capabilities = capabilities.session
    return (
        session_capabilities is not None
        and session_capabilities.config_options is not None
        and session_capabilities.config_options.boolean is not None
    )

supports_plan_content_updates()

Return whether the client supports unstable plan delta updates.

Source code in packages/adapters/pydantic-acp/src/pydantic_acp/session/state.py

def supports_plan_content_updates(self) -> bool:
    """Return whether the client supports unstable plan delta updates."""
    capabilities = self.client_capabilities
    return capabilities is None or capabilities.plan is not None

JsonValue = JsonPrimitive | list['JsonValue'] | dict[str, 'JsonValue'] module-attribute

RuntimeAgent = PydanticAgent[Any, Any] module-attribute

ACP Client Provider Bridge

AcpProvider(*, acp_agent, host_client=None, cwd='.', name='acp', base_url='acp://local', protocol_version=PROTOCOL_VERSION, client_capabilities=None, client_info=None, mcp_servers=None, prompt_renderer=None, history_mode='latest_user', enable_pydantic_acp_meta=None)

Bases: Provider[Agent]

Pydantic AI provider that treats an ACP agent as the model backend.

This is the inverse of the normal pydantic-acp server adapter. The server adapter exposes a pydantic_ai.Agent through ACP. AcpProvider consumes an existing ACP agent and makes it available to Pydantic AI as a provider/model pair, so application code can write ordinary Pydantic AI agents while delegating the underlying model turn to ACP.

The provider owns ACP protocol/session setup, model selection through the remote agent's model config option, host/client update capture, and prompt rendering. It deliberately remains a provider rather than an alternate agent framework: Pydantic AI still owns the outer agent run, result validation, usage accumulation, and history shape.

Create a new ACP provider.

Parameters:

Name	Type	Description	Default
acp_agent	Agent	The ACP agent to wrap as a Pydantic AI provider/model.	required
host_client	Client | None	Optional upstream :class:AcpClient to delegate real host operations to (file I/O, terminals, permissions, etc.).	None
cwd	str | Path	Working directory passed to the ACP session on creation.	'.'
name	str	Provider name reported via :attr:name. Defaults to "acp".	'acp'
base_url	str	Base URL reported via :attr:base_url. Defaults to "acp://local".	'acp://local'
protocol_version	int	ACP protocol version used during initialize. Defaults to :data:acp.PROTOCOL_VERSION.	PROTOCOL_VERSION
client_capabilities	ClientCapabilities | None	ACP client capabilities sent during initialize. Defaults to an empty :class:~acp.schema.ClientCapabilities.	None
client_info	Implementation | None	ACP implementation info sent during initialize. Defaults to a pydantic-acp-client stub.	None
mcp_servers	Sequence[Any] | None	Optional list of MCP servers forwarded to new_session.	None
prompt_renderer	AcpPromptRenderer | None	Custom callable that converts Pydantic AI messages into ACP prompt blocks. When None the built-in renderer is used.	None
history_mode	HistoryMode	Controls how previous messages are rendered into the ACP prompt. "latest_user" (default) sends only the latest user turn; "full" sends the entire conversation.	'latest_user'
enable_pydantic_acp_meta	bool | None	Enables the private pydantic_acp _meta extension used for ACP-backed Pydantic AI structured output. None auto-enables it only for ACP agents produced by this package; arbitrary ACP agents must not be trusted to implement this private contract.	None

Source code in packages/adapters/pydantic-acp/src/pydantic_acp/client.py

def __init__(
    self,
    *,
    acp_agent: AcpAgent,
    host_client: AcpClient | None = None,
    cwd: str | Path = ".",
    name: str = "acp",
    base_url: str = "acp://local",
    protocol_version: int = PROTOCOL_VERSION,
    client_capabilities: ClientCapabilities | None = None,
    client_info: Implementation | None = None,
    mcp_servers: Sequence[Any] | None = None,
    prompt_renderer: AcpPromptRenderer | None = None,
    history_mode: HistoryMode = "latest_user",
    enable_pydantic_acp_meta: bool | None = None,
) -> None:
    """Create a new ACP provider.

    Args:
        acp_agent: The ACP agent to wrap as a Pydantic AI provider/model.
        host_client: Optional upstream :class:`AcpClient` to delegate real
            host operations to (file I/O, terminals, permissions, etc.).
        cwd: Working directory passed to the ACP session on creation.
        name: Provider name reported via :attr:`name`. Defaults to
            ``"acp"``.
        base_url: Base URL reported via :attr:`base_url`. Defaults to
            ``"acp://local"``.
        protocol_version: ACP protocol version used during
            ``initialize``. Defaults to :data:`acp.PROTOCOL_VERSION`.
        client_capabilities: ACP client capabilities sent during
            ``initialize``. Defaults to an empty
            :class:`~acp.schema.ClientCapabilities`.
        client_info: ACP implementation info sent during ``initialize``.
            Defaults to a ``pydantic-acp-client`` stub.
        mcp_servers: Optional list of MCP servers forwarded to
            ``new_session``.
        prompt_renderer: Custom callable that converts Pydantic AI
            messages into ACP prompt blocks. When ``None`` the built-in
            renderer is used.
        history_mode: Controls how previous messages are rendered into
            the ACP prompt. ``"latest_user"`` (default) sends only the
            latest user turn; ``"full"`` sends the entire conversation.
        enable_pydantic_acp_meta: Enables the private ``pydantic_acp``
            ``_meta`` extension used for ACP-backed Pydantic AI structured
            output. ``None`` auto-enables it only for ACP agents produced
            by this package; arbitrary ACP agents must not be trusted to
            implement this private contract.

    """
    self._client = acp_agent

    self._host = AcpHostBridge(delegate=host_client)
    self._name = name
    self._base_url = base_url
    self._cwd = str(cwd)
    self._protocol_version = protocol_version
    self._client_capabilities = client_capabilities
    self._client_info = client_info or Implementation(
        name="pydantic-acp-client",
        version=__version__,
    )
    self._mcp_servers = list(mcp_servers or [])
    self._prompt_renderer = prompt_renderer
    self._history_mode = history_mode
    self._enable_pydantic_acp_meta = (
        _agent_supports_pydantic_acp_meta(acp_agent)
        if enable_pydantic_acp_meta is None
        else enable_pydantic_acp_meta
    )
    self._initialized = False
    self._session_id: str | None = None
    self._current_model_name: str | None = None
    self._model_config_option_available: bool | None = None
    self._session_lock: asyncio.Lock | None = None
    self._session_lock_loop: asyncio.AbstractEventLoop | None = None

    if hasattr(self._client, "on_connect"):
        self._client.on_connect(self._host)

host property

The ACP host bridge connected to the wrapped ACP agent.

updates property

All ACP updates recorded by the host bridge so far.

enable_pydantic_acp_meta property

Whether this provider opts into private pydantic_acp ACP metadata.

close() async

Close the wrapped ACP agent when it exposes an async-compatible close hook.

Source code in packages/adapters/pydantic-acp/src/pydantic_acp/client.py

async def close(self) -> None:
    """Close the wrapped ACP agent when it exposes an async-compatible close hook."""
    close = getattr(self._client, "close", None)
    if close is None:
        return
    result = close()
    if inspect.isawaitable(result):
        await result

model(model_name=None, *, settings=None, profile=None, history_mode=None)

Build an :class:AcpModel bound to this provider.

When model_name is None, the bridge does not call ACP session/set_config_option for "model" and leaves model selection to the wrapped agent's session default. The visible Pydantic AI model name remains "agent".

Source code in packages/adapters/pydantic-acp/src/pydantic_acp/client.py

def model(
    self,
    model_name: str | None = None,
    *,
    settings: ModelSettings | None = None,
    profile: ModelProfileSpec | None = None,
    history_mode: HistoryMode | None = None,
) -> AcpModel:
    """Build an :class:`AcpModel` bound to this provider.

    When ``model_name`` is ``None``, the bridge does not call ACP
    ``session/set_config_option`` for ``"model"`` and leaves model selection
    to the wrapped agent's session default. The visible Pydantic AI model
    name remains ``"agent"``.
    """
    return AcpModel(
        model_name=model_name,
        provider=self,
        settings=settings,
        profile=profile,
        history_mode=history_mode,
    )

render_prompt_blocks(messages, model_request_parameters, *, history_mode=None) async

Render Pydantic AI model messages into ACP prompt blocks.

Source code in packages/adapters/pydantic-acp/src/pydantic_acp/client.py

async def render_prompt_blocks(
    self,
    messages: Sequence[ModelMessage],
    model_request_parameters: ModelRequestParameters,
    *,
    history_mode: HistoryMode | None = None,
) -> list[AgentPromptBlock]:
    """Render Pydantic AI model messages into ACP prompt blocks."""
    if self._prompt_renderer is None:
        return _default_render_prompt_blocks(
            messages,
            model_request_parameters,
            history_mode=history_mode or self._history_mode,
        )
    rendered = self._prompt_renderer(messages, model_request_parameters)
    if inspect.isawaitable(rendered):
        rendered = await rendered
    return list(cast(Sequence[AgentPromptBlock], rendered))

request_prompt(*, model_name, prompt, model_request_parameters) async

Send one prompt turn to the ACP agent and collect its visible text.

Source code in packages/adapters/pydantic-acp/src/pydantic_acp/client.py

async def request_prompt(
    self,
    *,
    model_name: str | None,
    prompt: Sequence[AgentPromptBlock],
    model_request_parameters: ModelRequestParameters,
) -> _AcpPromptResult:
    """Send one prompt turn to the ACP agent and collect its visible text."""
    session_id = await self._ensure_session(model_name=model_name)
    start_index = self._host.snapshot_index()
    request_meta = None
    if self._enable_pydantic_acp_meta:
        request_meta = build_structured_output_request_meta(model_request_parameters)
    prompt_kwargs: dict[str, Any] = {
        "prompt": list(prompt),
        "session_id": session_id,
        "message_id": uuid4().hex,
    }
    if request_meta is not None:
        prompt_kwargs["_meta"] = request_meta
    prompt_response = await self._client.prompt(**prompt_kwargs)
    text = await self._agent_message_text_after_prompt(
        start_index,
        session_id=session_id,
        prompt_response=prompt_response,
    )

    response_meta = extract_field_meta(prompt_response)
    usage = _usage_from_acp(getattr(prompt_response, "usage", None))
    if not usage.has_values():
        usage = self._host.usage_update_since(start_index, session_id=session_id)
    stop_reason = getattr(prompt_response, "stop_reason", None) or getattr(
        prompt_response,
        "stopReason",
        None,
    )
    return _AcpPromptResult(
        text=text,
        usage=usage,
        stop_reason=stop_reason,
        session_id=session_id,
        structured_output=extract_structured_output(response_meta),
    )

AcpModel(model_name=None, *, provider, settings=None, profile=None, history_mode=None)

Bases: Model[Agent]

Pydantic AI Model backed by an ACP agent provider.

Source code in packages/adapters/pydantic-acp/src/pydantic_acp/client.py

def __init__(
    self,
    model_name: str | None = None,
    *,
    provider: AcpProvider,
    settings: ModelSettings | None = None,
    profile: ModelProfileSpec | None = None,
    history_mode: HistoryMode | None = None,
) -> None:
    self._model_name = model_name
    self._history_mode = history_mode
    self._provider = provider
    super().__init__(
        settings=settings,
        profile=profile or _profile_for_provider(provider),
    )

AcpHostBridge(*, delegate=None)

Minimal ACP host/client implementation used by :class:AcpProvider.

ACP agents send their visible output to a connected ACP client via session_update. Pydantic AI models, however, return a ModelResponse. This bridge is the seam between the two contracts: it records ACP updates so AcpModel can fold agent message chunks back into Pydantic AI response parts, while optionally delegating real host operations to an upstream ACP client supplied by the caller.

The bridge intentionally does not emulate a filesystem, terminal, approval UI, or extension namespace. When an ACP agent asks for such host operations and no delegate was supplied, the request fails explicitly instead of inventing host behavior that is not present.

Source code in packages/adapters/pydantic-acp/src/pydantic_acp/client.py

def __init__(self, *, delegate: AcpClient | None = None) -> None:
    self.delegate = delegate
    self.updates: list[AcpUpdateRecord] = []

snapshot_index()

Return an index that can later be used to read only new updates.

Source code in packages/adapters/pydantic-acp/src/pydantic_acp/client.py

def snapshot_index(self) -> int:
    """Return an index that can later be used to read only new updates."""
    return len(self.updates)

records_since(index, *, session_id=None)

Return recorded updates after index, optionally scoped to a session.

Source code in packages/adapters/pydantic-acp/src/pydantic_acp/client.py

def records_since(
    self,
    index: int,
    *,
    session_id: str | None = None,
) -> list[AcpUpdateRecord]:
    """Return recorded updates after ``index``, optionally scoped to a session."""
    records = self.updates[index:]
    if session_id is None:
        return list(records)
    return [record for record in records if record.session_id == session_id]

agent_message_text_since(index, *, session_id)

Concatenate ACP agent message chunks recorded after index.

Source code in packages/adapters/pydantic-acp/src/pydantic_acp/client.py

def agent_message_text_since(self, index: int, *, session_id: str) -> str:
    """Concatenate ACP agent message chunks recorded after ``index``."""
    text_parts: list[str] = []
    for record in self.records_since(index, session_id=session_id):
        update = record.update
        if not _is_agent_message_chunk(update):
            continue
        content = getattr(update, "content", None)
        text = getattr(content, "text", None)
        if isinstance(text, str):
            text_parts.append(text)
    return "".join(text_parts)

usage_update_since(index, *, session_id)

Collect the latest ACP usage update observed after index.

Source code in packages/adapters/pydantic-acp/src/pydantic_acp/client.py

def usage_update_since(self, index: int, *, session_id: str) -> RequestUsage:
    """Collect the latest ACP usage update observed after ``index``."""
    usage = RequestUsage()
    for record in self.records_since(index, session_id=session_id):
        update = record.update
        if not isinstance(update, UsageUpdate):
            continue
        usage = _usage_from_acp(getattr(update, "usage", None))
    return usage

session_update(session_id, update, **kwargs) async

Record an ACP update and optionally forward it to a real host client.

Source code in packages/adapters/pydantic-acp/src/pydantic_acp/client.py

async def session_update(
    self,
    session_id: str,
    update: Any,
    **kwargs: Any,
) -> None:
    """Record an ACP update and optionally forward it to a real host client."""
    self.updates.append(
        AcpUpdateRecord(
            session_id=session_id,
            update=update,
            source=kwargs.get("source"),
        ),
    )
    if self.delegate is not None and hasattr(self.delegate, "session_update"):
        await self._call_delegate(
            "session_update",
            session_id=session_id,
            update=update,
            **kwargs,
        )

on_connect(conn)

Forward reverse connections to a delegate host client when it supports them.

Source code in packages/adapters/pydantic-acp/src/pydantic_acp/client.py

def on_connect(self, conn: AcpAgent) -> None:
    """Forward reverse connections to a delegate host client when it supports them."""
    if self.delegate is not None and hasattr(self.delegate, "on_connect"):
        self.delegate.on_connect(conn)

Agent Source Classes And Protocols

AgentFactory

Bases: Protocol[AgentFactoryDepsT, AgentFactoryOutputDataT]

AgentSource

Bases: Protocol[AgentDepsT, OutputDataT]

StaticAgentSource(agent, deps=None) dataclass

Bases: Generic[AgentDepsT, OutputDataT]

FactoryAgentSource(factory) dataclass

Bases: Generic[AgentDepsT, OutputDataT]

Session Store Classes

SessionStore

Bases: Protocol

MemorySessionStore(_sessions=dict()) dataclass

FileSessionStore(root) dataclass

Provider State Classes And Protocols

ModelSelectionState(*, available_models, current_model_id, allow_any_model_id=False, enable_config_option=True, config_option_name='Model', config_option_description='Session-local model override.') dataclass

ModeState(*, modes, current_mode_id=None) dataclass

SessionModelsProvider

Bases: Protocol

SessionModesProvider

Bases: Protocol

ConfigOptionsProvider

Bases: Protocol

PlanProvider

Bases: Protocol

NativePlanPersistenceProvider

Bases: Protocol

ApprovalStateProvider

Bases: Protocol

ApprovalPolicy = Literal['allow', 'reject'] module-attribute

ApprovalPolicyStore

Bases: Protocol

SessionMetadataApprovalPolicyStore(metadata_key='approval_policies') dataclass

PermissionOptionSet(*, allow_once_name='Allow', reject_once_name='Deny', allow_always_name='Always Allow', reject_always_name='Always Deny') dataclass

Bridge Classes

CapabilityBridge

BufferedCapabilityBridge() dataclass

Bases: CapabilityBridge

PrepareToolsBridge(*, metadata_key='prepare_tools', default_mode_id, modes, mode_config_key='mode', plan_generation_config_id='plan_generation_type', plan_generation_config_name='Plan Generation', plan_generation_config_description='How plan mode records ACP plan state.', default_plan_generation_type='structured') dataclass

Bases: BufferedCapabilityBridge, Generic[AgentDepsT]

PrepareToolsMode(*, id, name, prepare_func, description=None, plan_mode=False, plan_tools=False) dataclass

Bases: Generic[AgentDepsT]

PrepareOutputToolsBridge(*, metadata_key='prepare_output_tools', default_mode_id, modes, mode_config_key='prepare_output_tools_mode') dataclass

Bases: BufferedCapabilityBridge, Generic[AgentDepsT]

PrepareOutputToolsMode(*, id, name, prepare_func, description=None) dataclass

Bases: Generic[AgentDepsT]

ThinkingBridge(*, config_id='thinking', config_name='Thinking Effort', config_description='Session-local thinking/reasoning effort.') dataclass

Bases: CapabilityBridge

HookBridge(metadata_key='hooks', hide_all=False, record_event_stream=True, record_model_requests=True, record_node_lifecycle=True, record_deferred_tool_calls=True, record_output_processing=True, record_output_validation=True, record_prepare_output_tools=True, record_prepare_tools=True, record_run_lifecycle=True, record_tool_execution=True, record_tool_validation=True) dataclass

Bases: BufferedCapabilityBridge

ExternalHookEventBridge(*, metadata_key='external_hooks', projection_map=HookProjectionMap(), emission_mode='paired') dataclass

Bases: BufferedCapabilityBridge

EventEmissionMode = Literal['paired', 'start_only'] module-attribute

HistoryProcessorBridge(metadata_key='history_processors', processor_names=list()) dataclass

Bases: BufferedCapabilityBridge

ThreadExecutorBridge(*, executor, metadata_key='thread_executor') dataclass

Bases: CapabilityBridge

ImageGenerationBridge(*, builtin=True, local=None, fallback_model=None, background=None, input_fidelity=None, moderation=None, output_compression=None, output_format=None, quality=None, size=None, aspect_ratio=None, tool_names=_DEFAULT_IMAGE_GENERATION_TOOL_NAMES, metadata_key='image_generation') dataclass

Bases: CapabilityBridge, Generic[AgentDepsT]

SetToolMetadataBridge(*, tools='all', metadata_key=None, **metadata) dataclass

Bases: CapabilityBridge, Generic[AgentDepsT]

Source code in packages/adapters/pydantic-acp/src/pydantic_acp/bridges/capability_support.py

def __init__(
    self,
    *,
    tools: ToolSelector[AgentDepsT] = "all",
    metadata_key: str | None = None,
    **metadata: JsonValue,
) -> None:
    self.tools = tools
    self.metadata_key = metadata_key
    self.metadata = dict(metadata)

IncludeToolReturnSchemasBridge(tools='all', metadata_key=None) dataclass

Bases: CapabilityBridge, Generic[AgentDepsT]

ToolsetBridge(*, toolset, metadata_key='toolset') dataclass

Bases: CapabilityBridge, Generic[AgentDepsT]

PrefixToolsBridge(*, wrapped, prefix, metadata_key='prefix_tools') dataclass

Bases: CapabilityBridge, Generic[AgentDepsT]

WebSearchBridge(*, builtin=True, local=None, search_context_size=None, user_location=None, blocked_domains=None, allowed_domains=None, max_uses=None, tool_names=_DEFAULT_WEB_SEARCH_TOOL_NAMES, metadata_key='web_search') dataclass

Bases: CapabilityBridge, Generic[AgentDepsT]

WebFetchBridge(*, builtin=True, local=None, allowed_domains=None, blocked_domains=None, max_uses=None, enable_citations=None, max_content_tokens=None, tool_names=_DEFAULT_WEB_FETCH_TOOL_NAMES, metadata_key='web_fetch') dataclass

Bases: CapabilityBridge, Generic[AgentDepsT]

McpCapabilityBridge(*, url, builtin=True, local=None, id=None, authorization_token=None, headers=None, allowed_tools=None, description=None, tool_name_prefixes=_DEFAULT_MCP_TOOL_NAME_PREFIXES, metadata_key='mcp_capability') dataclass

Bases: CapabilityBridge, Generic[AgentDepsT]

SessionMcpBridge(*, metadata_key='session_mcp', include_instructions=True, include_return_schema=None, cache_tools=True, cache_resources=True, cache_prompts=True, tool_error_behavior='retry', max_retries=None, allowed_tools=None, tool_name_prefixes=frozenset(), toolset_id_prefix='acp-session-mcp', advertise_http=True, advertise_sse=True) dataclass

Bases: CapabilityBridge

Attach ACP client-provided MCP servers to the Pydantic AI agent run.

ACP clients may pass MCP server definitions during session/new, session/load, session/fork, or session/resume. The adapter persists those definitions on AcpSessionContext.mcp_servers; this bridge turns them into a Pydantic AI MCPToolset capability for the active session.

OpenAICompactionBridge(*, message_count_threshold=None, trigger=None, instructions=None, metadata_key='openai_compaction') dataclass

Bases: BufferedCapabilityBridge, Generic[AgentDepsT]

AnthropicCompactionBridge(*, token_threshold=150000, instructions=None, pause_after_compaction=False, metadata_key='anthropic_compaction') dataclass

Bases: CapabilityBridge, Generic[AgentDepsT]

McpBridge(*, metadata_key='mcp', approval_policy_scope='tool', config_options=list(), servers=list(), tools=list()) dataclass

Bases: CapabilityBridge

McpServerDefinition(*, server_id, name, transport, url=None, description=None, tool_prefix=None) dataclass

McpToolDefinition(*, tool_name, server_id, kind='execute') dataclass

Hook Introspection Helpers

RegisteredHookInfo(*, event_id, hook_name, tool_filters) dataclass

list_agent_hooks(agent)

Source code in packages/adapters/pydantic-acp/src/pydantic_acp/runtime/hook_introspection.py

def list_agent_hooks(agent: PydanticAgent[Any, Any]) -> list[RegisteredHookInfo]:
    resolved_root_capability = _root_capability(agent)
    if resolved_root_capability is None:
        return []
    hook_infos: list[RegisteredHookInfo] = []
    for hooks in _iter_hooks(resolved_root_capability):
        registry = hook_registry(hooks)
        if registry is None:
            continue
        for registry_key, entries in registry.items():
            event_id = _INTERNAL_EVENT_NAMES.get(registry_key, registry_key)
            for entry in entries:
                func = entry_func(entry)
                if not callable(func):
                    continue
                if getattr(func, "__module__", "") == _SKIPPED_HOOK_MODULE:
                    continue
                hook_infos.append(
                    RegisteredHookInfo(
                        event_id=event_id,
                        hook_name=getattr(func, "__name__", "") or event_id,
                        tool_filters=_tool_filters(entry),
                    ),
                )
    return sorted(
        hook_infos,
        key=lambda hook_info: (
            hook_info.event_id,
            hook_info.hook_name,
            hook_info.tool_filters,
        ),
    )

Projection Classes

FileSystemProjectionMap(*, write_tool_names=frozenset(), read_tool_names=frozenset(), bash_tool_names=frozenset(), search_tool_names=frozenset(), default_write_tool=None, default_read_tool=None, default_bash_tool=None, default_search_tool=None, path_arg=None, content_arg=None, old_text_arg=None, command_arg=None, terminal_id_arg=None, search_path_arg=None, search_pattern_arg=None, render_search_results_as_tree=False, hide_dot_directories_in_tree=True, tree_root_label=None) dataclass

WebToolProjectionMap(*, search_tool_names=_DEFAULT_SEARCH_TOOL_NAMES, fetch_tool_names=_DEFAULT_FETCH_TOOL_NAMES) dataclass

BuiltinToolProjectionMap(*, web_projection_map=WebToolProjectionMap(), image_generation_tool_names=_DEFAULT_IMAGE_GENERATION_TOOL_NAMES, mcp_tool_name_prefixes=_DEFAULT_MCP_TOOL_NAME_PREFIXES) dataclass

CompositeProjectionMap(*, maps) dataclass

ProjectionAwareToolClassifier(*, base_classifier, projection_maps) dataclass

Approval Presentation

PermissionRequestContext(*, session, tool_call, raw_input, cwd, classifier, projection_map=None) dataclass

PermissionToolCallBuilder

Bases: Protocol

DefaultPermissionToolCallBuilder(*, status='pending') dataclass

NativeApprovalBridge(*, enable_persistent_choices=False, tool_call_builder=DefaultPermissionToolCallBuilder(), policy_store=SessionMetadataApprovalPolicyStore(), option_set=PermissionOptionSet()) dataclass

ProjectionAwareApprovalBridge

Bases: Protocol

supports_projection_aware_approval_bridge(value)

Source code in packages/adapters/pydantic-acp/src/pydantic_acp/approvals.py

def supports_projection_aware_approval_bridge(
    value: object,
) -> TypeIs[ProjectionAwareApprovalBridge]:
    resolver = getattr(value, "resolve_deferred_approvals", None)
    if not callable(resolver):
        return False
    try:
        resolver_signature = signature(resolver)
    except (TypeError, ValueError):
        return False
    parameters = resolver_signature.parameters
    if "projection_map" in parameters:
        return True
    return any(parameter.kind is Parameter.VAR_KEYWORD for parameter in parameters.values())

Slash Commands

SlashCommandRequest(*, name, argument, raw_prompt, session, agent) dataclass

SlashCommandResult(*, text=None, updates=(), stop_reason='end_turn', handled=True, refresh_session_surface=True) dataclass

SlashCommandProvider

Bases: Protocol

StaticSlashCommand(*, command, handler) dataclass

StaticSlashCommandProvider(*, commands) dataclass

SlashCommandHandler = Callable[[SlashCommandRequest], SlashCommandResult | None | Awaitable[SlashCommandResult | None]] module-attribute

Projection Helpers

truncate_text(text, *, limit, marker=DEFAULT_TEXT_TRUNCATION_MARKER)

Source code in packages/adapters/pydantic-acp/src/pydantic_acp/_projection_text.py

def truncate_text(
    text: str,
    *,
    limit: int,
    marker: str = DEFAULT_TEXT_TRUNCATION_MARKER,
) -> str:
    if limit <= 0:
        return ""
    if len(text) <= limit:
        return text
    if limit <= len(marker):
        return f"{text[:limit]}{marker}"
    return f"{text[: limit - len(marker)]}{marker}"

truncate_lines(lines, *, max_lines, truncation_line='... [truncated]')

Source code in packages/adapters/pydantic-acp/src/pydantic_acp/_projection_text.py

def truncate_lines(
    lines: Sequence[str],
    *,
    max_lines: int,
    truncation_line: str = "... [truncated]",
) -> list[str]:
    if max_lines <= 0:
        return []
    materialized = list(lines)
    if len(materialized) <= max_lines:
        return materialized
    if max_lines == 1:
        return [truncation_line]
    return [*materialized[: max_lines - 1], truncation_line]

single_line_summary(text, *, limit)

Source code in packages/adapters/pydantic-acp/src/pydantic_acp/_projection_text.py

def single_line_summary(
    text: str,
    *,
    limit: int,
) -> str:
    normalized = " ".join(text.split())
    if len(normalized) <= limit:
        return normalized
    return f"{normalized[:limit].rstrip()}..."

format_code_block(text, *, language=None, limit=None)

Source code in packages/adapters/pydantic-acp/src/pydantic_acp/_projection_text.py

def format_code_block(
    text: str,
    *,
    language: str | None = None,
    limit: int | None = None,
) -> str:
    body = truncate_text(text, limit=limit) if limit is not None else text
    if language is None:
        return f"```\n{body}\n```"
    return f"```{language}\n{body}\n```"

format_diff_preview(path, old_text, new_text, *, context_lines=3, max_lines=40, include_path_header=True, include_diff_headers=False)

Source code in packages/adapters/pydantic-acp/src/pydantic_acp/_projection_text.py

def format_diff_preview(
    path: str | Path,
    old_text: str,
    new_text: str,
    *,
    context_lines: int = 3,
    max_lines: int = 40,
    include_path_header: bool = True,
    include_diff_headers: bool = False,
) -> str:
    diff_lines = list(
        unified_diff(
            old_text.strip().splitlines(),
            new_text.strip().splitlines(),
            lineterm="",
            n=context_lines,
        ),
    )
    if not include_diff_headers:
        diff_lines = [
            line
            for line in diff_lines
            if not line.startswith("--- ") and not line.startswith("+++ ")
        ]
    if not diff_lines:
        diff_lines = ["(no visible changes)"]
    body_lines: list[str] = []
    if include_path_header:
        body_lines.append(f"# {path}")
    body_lines.extend(truncate_lines(diff_lines, max_lines=max_lines))
    return "\n".join(body_lines)

format_terminal_status(*, exit_code, signal)

Source code in packages/adapters/pydantic-acp/src/pydantic_acp/_projection_text.py

def format_terminal_status(
    *,
    exit_code: int | None,
    signal: str | None,
) -> str:
    if signal is not None:
        return f"cancelled ({signal})"
    if exit_code is None:
        return "running"
    if exit_code == 0:
        return "ok (0)"
    return f"fail ({exit_code})"

caution_for_path(path, *, session_cwd, workspace_root=None, access_policy=None)

Source code in packages/adapters/pydantic-acp/src/pydantic_acp/_projection_risk.py

def caution_for_path(
    path: str | Path,
    *,
    session_cwd: Path,
    workspace_root: Path | None = None,
    access_policy: HostAccessPolicy | None = None,
) -> str | None:
    policy = access_policy or HostAccessPolicy()
    evaluation = policy.evaluate_path(
        path,
        session_cwd=session_cwd,
        workspace_root=workspace_root,
    )
    if not evaluation.has_risks:
        return None
    return evaluation.message

caution_for_command(command, *, args=None, cwd=None, session_cwd, workspace_root=None, access_policy=None)

Source code in packages/adapters/pydantic-acp/src/pydantic_acp/_projection_risk.py

def caution_for_command(
    command: str,
    *,
    args: Sequence[str] | None = None,
    cwd: str | Path | None = None,
    session_cwd: Path,
    workspace_root: Path | None = None,
    access_policy: HostAccessPolicy | None = None,
) -> str | None:
    policy = access_policy or HostAccessPolicy()
    evaluation = policy.evaluate_command(
        command,
        args=args,
        cwd=cwd,
        session_cwd=session_cwd,
        workspace_root=workspace_root,
    )
    if not evaluation.has_risks:
        return None
    return evaluation.message

Host Backend Classes

ClientHostContext(*, client, session, filesystem, terminal, access_policy=None, workspace_root=None) dataclass

ClientFilesystemBackend(*, client, session, access_policy=None, workspace_root=None) dataclass

ClientTerminalBackend(*, client, session, access_policy=None, workspace_root=None) dataclass

Testing Helpers

BlackBoxHarness(*, adapter, client=RecordingACPClient(), last_session_id=None) dataclass

RecordingACPClient(*, updates=list(), permission_option_ids=list(), permission_option_names=list(), permission_responses=list(), read_calls=list(), write_calls=list(), create_calls=list(), output_calls=list(), release_calls=list(), wait_calls=list(), kill_calls=list(), write_response=WriteTextFileResponse(), release_response=ReleaseTerminalResponse(), kill_response=KillTerminalResponse(), wait_response=(lambda: WaitForTerminalExitResponse(exit_code=0))(), terminal_output_response=(lambda: TerminalOutputResponse(output='terminal-output', truncated=False))()) dataclass