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,
) -> AcpAgent:
    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

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, hook_projection_map=HookProjectionMap(), models_provider=None, modes_provider=None, native_plan_persistence_provider=None, plan_provider=None, replay_history_on_load=True, available_models=list(), session_store=MemorySessionStore(), output_serializer=DefaultOutputSerializer(), projection_maps=tuple(), tool_classifier=DefaultToolClassifier()) dataclass

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

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

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

RuntimeAgent = PydanticAgent[Any, Any] module-attribute

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

Bridge Classes

CapabilityBridge

BufferedCapabilityBridge() dataclass

Bases: CapabilityBridge

PrepareToolsBridge(*, metadata_key='prepare_tools', default_mode_id, modes, mode_config_key='mode') dataclass

Bases: BufferedCapabilityBridge, Generic[AgentDepsT]

PrepareToolsMode(*, id, name, prepare_func, description=None, plan_mode=False, plan_tools=False) 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_prepare_tools=True, record_run_lifecycle=True, record_tool_execution=True, record_tool_validation=True) dataclass

Bases: BufferedCapabilityBridge

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

Bases: BufferedCapabilityBridge

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]:
    root_capability = _root_capability(agent)
    if root_capability is None:
        return []
    hook_infos: list[RegisteredHookInfo] = []
    for hooks in _iter_hooks(root_capability):
        registry = getattr(hooks, "_registry", None)
        if not isinstance(registry, dict):
            continue
        for registry_key, entries in registry.items():
            if not isinstance(registry_key, str) or not isinstance(entries, list):
                continue
            event_id = _INTERNAL_EVENT_NAMES.get(registry_key, registry_key)
            for entry in entries:
                if not isinstance(entry, _HookEntry):
                    continue
                func = getattr(entry, "func", None)
                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(), default_write_tool=None, default_read_tool=None, default_bash_tool=None, path_arg=None, content_arg=None, old_text_arg=None, command_arg=None, terminal_id_arg=None) dataclass

CompositeProjectionMap(*, maps) dataclass

Host Backend Classes

ClientHostContext(*, client, session, filesystem, terminal) dataclass

ClientFilesystemBackend(*, client, session) dataclass

ClientTerminalBackend(*, client, session) dataclass