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, 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_provider=None, 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, 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

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]

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