server

"""DeepAgents ACP server implementation."""

from future import annotations

import asyncio import uuid from typing import Any, Literal

from acp import ( Agent, AgentSideConnection, PROTOCOL_VERSION, stdio_streams, ) from acp.schema import ( AgentMessageChunk, InitializeRequest, InitializeResponse, NewSessionRequest, NewSessionResponse, PromptRequest, PromptResponse, SessionNotification, TextContentBlock, Implementation, AgentThoughtChunk, ToolCallProgress, ContentToolCallContent, LoadSessionResponse, SetSessionModeResponse, SetSessionModelResponse, CancelNotification, LoadSessionRequest, SetSessionModeRequest, SetSessionModelRequest, AgentPlanUpdate, PlanEntry, PermissionOption, RequestPermissionRequest, AllowedOutcome, DeniedOutcome, ToolCall as ACPToolCall, ) from deepagents import create_deep_agent from langchain_anthropic import ChatAnthropic from langchain_core.messages import AIMessage, AIMessageChunk, ToolMessage from langchain_core.messages.content import ToolCall from langchain_core.tools import tool from langgraph.checkpoint.memory import InMemorySaver from langgraph.graph.state import CompiledStateGraph from langgraph.types import Command, Interrupt

class DeepagentsACP(Agent): """ACP Agent implementation wrapping deepagents."""

def __init__(
    self,
    connection: AgentSideConnection,
    agent_graph: CompiledStateGraph,
) -> None:
    """Initialize the DeepAgents agent.

    Args:
        connection: The ACP connection for communicating with the client
        agent_graph: A compiled LangGraph StateGraph (output of create_deep_agent)
    """
    self._connection = connection
    self._agent_graph = agent_graph
    self._sessions: dict[str, dict[str, Any]] = {}
    # Track tool calls by ID for matching with ToolMessages
    # Maps tool_call_id -> ToolCall TypedDict
    self._tool_calls: dict[str, ToolCall] = {}

async def initialize(
    self,
    params: InitializeRequest,
) -> InitializeResponse:
    """Initialize the agent and return capabilities."""
    return InitializeResponse(
        protocolVersion=PROTOCOL_VERSION,
        agentInfo=Implementation(
            name="DeepAgents ACP Server",
            version="0.1.0",
            title="DeepAgents ACP Server",
        ),
    )

async def newSession(
    self,
    params: NewSessionRequest,
) -> NewSessionResponse:
    """Create a new session with a deepagents instance."""
    session_id = str(uuid.uuid4())
    # Store session state with the shared agent graph
    self._sessions[session_id] = {
        "agent": self._agent_graph,
        "thread_id": str(uuid.uuid4()),
    }

    return NewSessionResponse(sessionId=session_id)

async def _handle_ai_message_chunk(
    self,
    params: PromptRequest,
    message: AIMessageChunk,
) -> None:
    """Handle an AIMessageChunk and send appropriate notifications.

    Args:
        params: The prompt request parameters
        message: An AIMessageChunk from the streaming response

    Note:
        According to LangChain's content block types, message.content_blocks
        returns a list of ContentBlock unions. Each block is a TypedDict with
        a "type" field that discriminates the block type:
        - TextContentBlock: type="text", has "text" field
        - ReasoningContentBlock: type="reasoning", has "reasoning" field
        - ToolCallChunk: type="tool_call_chunk"
        - And many others (image, audio, video, etc.)
    """
    for block in message.content_blocks:
        # All content blocks have a "type" field for discrimination
        block_type = block.get("type")

        if block_type == "text":
            # TextContentBlock has a required "text" field
            text = block.get("text", "")
            if not text:  # Only yield non-empty text
                continue
            await self._connection.sessionUpdate(
                SessionNotification(
                    update=AgentMessageChunk(
                        content=TextContentBlock(text=text, type="text"),
                        sessionUpdate="agent_message_chunk",
                    ),
                    sessionId=params.sessionId,
                )
            )
        elif block_type == "reasoning":
            # ReasoningContentBlock has a "reasoning" field (NotRequired)
            reasoning = block.get("reasoning", "")
            if not reasoning:
                continue