langchain_acp API

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

Functions

create_acp_agent(graph=None, *, graph_factory=None, graph_source=None, config=None, event_projection_maps=None, projection_maps=None)

Source code in packages/adapters/langchain-acp/src/langchain_acp/runtime/server.py

def create_acp_agent(
    graph: Any | None = None,
    *,
    graph_factory: GraphFactory | None = None,
    graph_source: GraphSource | None = None,
    config: AdapterConfig | None = None,
    event_projection_maps: list[EventProjectionMap] | None = None,
    projection_maps: list[ProjectionMap] | None = None,
) -> AcpAgent:
    resolved_source = _resolve_graph_source(
        graph=graph,
        graph_factory=graph_factory,
        graph_source=graph_source,
    )
    resolved_config = _resolve_config(
        config=config,
        event_projection_maps=event_projection_maps,
        graph_name=getattr(graph, "name", None),
        projection_maps=projection_maps,
    )
    return LangChainAcpAgent(resolved_source, config=resolved_config)

run_acp(graph=None, *, graph_factory=None, graph_source=None, config=None, event_projection_maps=None, projection_maps=None)

Source code in packages/adapters/langchain-acp/src/langchain_acp/runtime/server.py

def run_acp(
    graph: Any | None = None,
    *,
    graph_factory: GraphFactory | None = None,
    graph_source: GraphSource | None = None,
    config: AdapterConfig | None = None,
    event_projection_maps: list[EventProjectionMap] | None = None,
    projection_maps: list[ProjectionMap] | None = None,
) -> None:
    adapter = create_acp_agent(
        graph=graph,
        graph_factory=graph_factory,
        graph_source=graph_source,
        config=config,
        event_projection_maps=event_projection_maps,
        projection_maps=projection_maps,
    )
    asyncio.run(run_agent(adapter))

compose_projection_maps(projection_maps)

Source code in packages/adapters/langchain-acp/src/langchain_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))

compose_event_projection_maps(projection_maps)

Source code in packages/adapters/langchain-acp/src/langchain_acp/event_projection.py

def compose_event_projection_maps(
    projection_maps: Sequence[EventProjectionMap] | None,
) -> EventProjectionMap | 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 CompositeEventProjectionMap(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, approval_bridge=NativeApprovalBridge(), available_models=list(), available_modes=list(), capability_bridges=tuple(), config_options_provider=None, default_model_id=None, default_mode_id=None, default_plan_generation_type='structured', enable_plan_progress_tools=False, event_projection_maps=tuple(), models_provider=None, modes_provider=None, native_plan_additional_instructions=None, native_plan_persistence_provider=None, output_serializer=DefaultOutputSerializer(), plan_mode_id=None, plan_provider=None, projection_maps=tuple(), replay_history_on_load=True, session_store=MemorySessionStore(), tool_classifier=DefaultToolClassifier()) dataclass

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

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

TaskPlan

Bases: BaseModel

Graph Source Classes And Protocols

CompiledAgentGraph = CompiledStateGraph[Any, Any, Any, Any] module-attribute

GraphFactory

Bases: Protocol

GraphSource

Bases: Protocol

StaticGraphSource(graph) dataclass

FactoryGraphSource(factory) dataclass

Session Store Classes

SessionStore

Bases: Protocol

MemorySessionStore(_sessions=dict()) dataclass

FileSessionStore(root) dataclass

Provider State Classes And Protocols

ConfigOption = SessionConfigOptionSelect | SessionConfigOptionBoolean module-attribute

ConfigOptionsProvider

Bases: Protocol

ModelSelectionState(*, available_models, current_model_id, allow_any_model_id=False, enable_config_option=True, config_option_name='Model') dataclass

ModeState(*, modes, current_mode_id=None, enable_config_option=True, config_option_name='Mode') dataclass

NativePlanPersistenceProvider

Bases: Protocol

PlanProvider

Bases: Protocol

SessionModelsProvider

Bases: Protocol

SessionModesProvider

Bases: Protocol

Bridge Classes

CapabilityBridge

BufferedCapabilityBridge() dataclass

Bases: CapabilityBridge

ConfigOptionsBridge(*, provider) dataclass

Bases: CapabilityBridge

ModelSelectionBridge(*, available_models=(), default_model_id=None, provider=None) dataclass

Bases: CapabilityBridge

ModeSelectionBridge(*, available_modes=(), default_mode_id=None, provider=None) dataclass

Bases: CapabilityBridge

ToolSurfaceBridge(*, tool_kinds=dict(), approval_policy_keys=dict()) dataclass

Bases: CapabilityBridge

DeepAgentsCompatibilityBridge(*, metadata_key='deepagents') dataclass

Bases: CapabilityBridge

GraphBridgeBuilder(*, base_classifier, bridges) dataclass

GraphBuildContributions(*, interrupt_configuration=dict(), metadata=dict(), middleware=(), response_format=None, system_prompt_parts=(), tools=()) dataclass

Plan Helpers

NativePlanGeneration = TaskPlan module-attribute

PlanGenerationType = Literal['tools', 'structured'] module-attribute

acp_get_plan()

Return the saved plan and numbered entries.

Source code in packages/adapters/langchain-acp/src/langchain_acp/plan.py

def acp_get_plan() -> str:
    """Return the saved plan and numbered entries."""
    context = _active_plan_context()
    if context is None:
        return "No active ACP session is bound."
    return context.runtime.format_native_plan(context.session)

acp_mark_plan_done(index) async

Mark a single plan entry completed by its 1-based index.

Source code in packages/adapters/langchain-acp/src/langchain_acp/plan.py

async def acp_mark_plan_done(index: int) -> str:
    """Mark a single plan entry completed by its 1-based index."""
    context = _active_plan_context()
    if context is None:
        return "No active ACP session is bound."
    updated_entry = await context.runtime.update_native_plan_entry(
        context.session,
        index=index,
        status="completed",
    )
    return f"Marked plan entry {index} as completed: {updated_entry.content}"

acp_set_plan(entries, plan_md=None) async

Replace the current ACP-owned plan state.

Source code in packages/adapters/langchain-acp/src/langchain_acp/plan.py

async def acp_set_plan(entries: list[PlanEntry], plan_md: str | None = None) -> str:
    """Replace the current ACP-owned plan state."""
    context = _active_plan_context()
    if context is None:
        return "No active ACP session is bound."
    await context.runtime.persist_native_plan_state(
        context.session,
        entries=entries,
        plan_markdown=plan_md,
    )
    return f"Recorded {len(entries)} plan entries."

acp_update_plan_entry(index, status=None, content=None, priority=None) async

Update a single plan entry by its 1-based index.

Source code in packages/adapters/langchain-acp/src/langchain_acp/plan.py

async def acp_update_plan_entry(
    index: int,
    status: str | None = None,
    content: str | None = None,
    priority: str | None = None,
) -> str:
    """Update a single plan entry by its 1-based index."""
    context = _active_plan_context()
    if context is None:
        return "No active ACP session is bound."
    updated_entry = await context.runtime.update_native_plan_entry(
        context.session,
        index=index,
        status=status,
        content=content,
        priority=priority,
    )
    return (
        f"Updated plan entry {index}: "
        f"[{updated_entry.status}] ({updated_entry.priority}) {updated_entry.content}"
    )

native_plan_tools()

Source code in packages/adapters/langchain-acp/src/langchain_acp/plan.py

def native_plan_tools() -> tuple[object, ...]:
    return (
        acp_get_plan,
        acp_set_plan,
        acp_update_plan_entry,
        acp_mark_plan_done,
    )

Projection Classes

FileSystemProjectionMap(*, write_tool_names=frozenset(), read_tool_names=frozenset(), search_tool_names=frozenset(), execute_tool_names=frozenset()) dataclass

DeepAgentsProjectionMap(*, base=(lambda: FileSystemProjectionMap(read_tool_names=(frozenset({'read_file'})), write_tool_names=(frozenset({'edit_file', 'write_file'})), search_tool_names=(frozenset({'glob', 'grep', 'ls'})), execute_tool_names=(frozenset({'execute'}))))()) dataclass

CompositeProjectionMap(*, maps) dataclass

StructuredEventProjectionMap(*, event_keys=_EVENT_KEYS) dataclass

CompositeEventProjectionMap(*, maps) dataclass

Projection Helpers

build_tool_start_update(*, tool_call_id, tool_name, classifier, raw_input, cwd, projection_map)

Source code in packages/adapters/langchain-acp/src/langchain_acp/projection.py

def build_tool_start_update(
    *,
    tool_call_id: str,
    tool_name: str,
    classifier: ToolClassifier,
    raw_input: Any,
    cwd: Path | None,
    projection_map: ProjectionMap | None,
) -> ToolCallStart:
    projected = (
        projection_map.project_start(tool_name, cwd=cwd, raw_input=raw_input)
        if projection_map
        else None
    )
    return ToolCallStart(
        session_update="tool_call",
        tool_call_id=tool_call_id,
        title=(projected.title if projected is not None else None) or tool_name,
        kind=classifier.classify(tool_name, raw_input),
        locations=(
            projected.locations
            if projected is not None and projected.locations is not None
            else None
        )
        or extract_tool_call_locations(raw_input)
        or None,
        raw_input=raw_input,
        status=(projected.status if projected is not None else None) or "in_progress",
        content=projected.content if projected is not None else None,
    )

build_tool_progress_update(*, tool_call_id, tool_name, classifier, raw_input, raw_output, serialized_output, cwd, projection_map, status)

Source code in packages/adapters/langchain-acp/src/langchain_acp/projection.py

def build_tool_progress_update(
    *,
    tool_call_id: str,
    tool_name: str,
    classifier: ToolClassifier,
    raw_input: Any,
    raw_output: Any,
    serialized_output: str,
    cwd: Path | None,
    projection_map: ProjectionMap | None,
    status: ToolCallStatus,
) -> ToolCallProgress:
    projected = (
        projection_map.project_progress(
            tool_name,
            cwd=cwd,
            raw_input=raw_input,
            raw_output=raw_output,
            serialized_output=serialized_output,
            status=status,
        )
        if projection_map
        else None
    )
    return ToolCallProgress(
        session_update="tool_call_update",
        tool_call_id=tool_call_id,
        title=projected.title if projected is not None else None,
        kind=classifier.classify(tool_name, raw_input),
        locations=(
            projected.locations
            if projected is not None and projected.locations is not None
            else None
        )
        or extract_tool_call_locations(raw_input)
        or None,
        raw_input=raw_input,
        raw_output=raw_output,
        status=(projected.status if projected is not None else None) or status,
        content=(
            projected.content
            if projected is not None and projected.content is not None
            else (
                [ContentToolCallContent(type="content", content=_text_block(serialized_output))]
                if serialized_output
                else None
            )
        ),
    )

extract_tool_call_locations(raw_input)

Source code in packages/adapters/langchain-acp/src/langchain_acp/projection.py

def extract_tool_call_locations(raw_input: Any) -> list[ToolCallLocation]:
    if not _is_string_keyed_object_dict(raw_input):
        return []
    path = _first_string(raw_input, _PATH_KEYS)
    if path is None:
        return []
    return [ToolCallLocation(path=path)]