server

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

            await self._connection.sessionUpdate(
                SessionNotification(
                    update=AgentThoughtChunk(
                        content=TextContentBlock(text=reasoning, type="text"),
                        sessionUpdate="agent_thought_chunk",
                    ),
                    sessionId=params.sessionId,
                )
            )

async def _handle_completed_tool_calls(
    self,
    params: PromptRequest,
    message: AIMessage,
) -> None:
    """Handle completed tool calls from an AIMessage and send notifications.

    Args:
        params: The prompt request parameters
        message: An AIMessage containing tool_calls

    Note:
        According to LangChain's AIMessage type:
        - message.tool_calls: list[ToolCall] where ToolCall is a TypedDict with:
          - name: str (required)
          - args: dict[str, Any] (required)
          - id: str | None (required field, but can be None)
          - type: Literal["tool_call"] (optional, NotRequired)
    """
    # Use direct attribute access - tool_calls is a defined field on AIMessage
    if not message.tool_calls:
        return

    for tool_call in message.tool_calls:
        # Access TypedDict fields directly (they're required fields)
        tool_call_id = tool_call["id"]  # str | None
        tool_name = tool_call["name"]  # str
        tool_args = tool_call["args"]  # dict[str, Any]

        # Skip tool calls without an ID (shouldn't happen in practice)
        if tool_call_id is None:
            continue

        # Skip todo tool calls as they're handled separately
        if tool_name == "todo":
            raise NotImplementedError("TODO tool call handling not implemented yet")

        # Send tool call progress update showing the tool is running
        await self._connection.sessionUpdate(
            SessionNotification(
                update=ToolCallProgress(
                    sessionUpdate="tool_call_update",
                    toolCallId=tool_call_id,
                    title=tool_name,
                    rawInput=tool_args,
                    status="pending",
                ),
                sessionId=params.sessionId,
            )
        )

        # Store the tool call for later matching with ToolMessage
        self._tool_calls[tool_call_id] = tool_call

async def _handle_tool_message(
    self,
    params: PromptRequest,
    tool_call: ToolCall,
    message: ToolMessage,
) -> None:
    """Handle a ToolMessage and send appropriate notifications.

    Args:
        params: The prompt request parameters
        tool_call: The original ToolCall that this message is responding to
        message: A ToolMessage containing the tool execution result

    Note:
        According to LangChain's ToolMessage type (inherits from BaseMessage):
        - message.content: str | list[str | dict] (from BaseMessage)
        - message.tool_call_id: str (specific to ToolMessage)
        - message.status: str | None (e.g., "error" for failed tool calls)
    """
    # Determine status based on message status or content
    status: Literal["completed", "failed"] = "completed"
    if hasattr(message, "status") and message.status == "error":
        status = "failed"

    # Build content blocks if message has content
    content_blocks = []
    for content_block in message.content_blocks:
        if content_block.get("type") == "text":
            text = content_block.get("text", "")
            if text:
                content_blocks.append(
                    ContentToolCallContent(
                        type="content",
                        content=TextContentBlock(text=text, type="text"),
                    )
                )
    # Send tool call progress update with the result
    await self._connection.sessionUpdate(
        SessionNotification(
            update=ToolCallProgress(
                sessionUpdate="tool_call_update",
                toolCallId=message.tool_call_id,
                title=tool_call["name"],
                content=content_blocks,
                rawOutput=message.content,
                status=status,
            ),
            sessionId=params.sessionId,
        )
    )

async def _handle_todo_update(
    self,
    params: PromptRequest,
    todos: list[dict[str, Any]],
) -> None:
    """Handle todo list updates from the tools node.

    Args:
        params: The prompt request parameters
        todos: List of todo dictionaries with 'content' and 'status' fields

    Note:
        Todos come from the deepagents graph's write_todos tool and have the structure:
        [{'content': 'Task description', 'status': 'pending'|'in_progress'|'completed'}, ...]
    """
    # Convert todos to PlanEntry objects
    entries = []
    for todo in todos:
        # Extract fields from todo dict
        content = todo.get("content", "")
        status = todo.get("status", "pending")

        # Validate and cast status to PlanEntryStatus
        if status not in ("pending", "in_progress", "completed"):
            status = "pending"

        # Create PlanEntry with default priority of "medium"
        entry = PlanEntry(
            content=content,
            status=status,  # type: ignore
            priority="medium",
        )
        entries.append(entry)

    # Send plan update notification
    await self._connection.sessionUpdate(
        SessionNotification(
            update=AgentPlanUpdate(
                sessionUpdate="plan",
                entries=entries,
            ),
            sessionId=params.sessionId,
        )
    )

async def _handle_interrupt(
    self,
    params: PromptRequest,
    interrupt: Interrupt,
) -> list[dict[str, Any]]:
    """Handle a LangGraph interrupt and request permission from the client.

    Args:
        params: The prompt request parameters
        interrupt: The interrupt from LangGraph containing action_requests and review_configs

    Returns:
        List of decisions to pass to Command(resume={...})

    Note:
        The interrupt.value contains:
        - action_requests: [{'name': str, 'args': dict, 'description': str}, ...]
        - review_configs: [{'action_name': str, 'allowed_decisions': list[str]}, ...]
    """
    interrupt_data = interrupt.value
    action_requests = interrupt_data.get("action_requests", [])
    review_configs = interrupt_data.get("review_configs", [])

    # Create a mapping of action names to their allowed decisions
    allowed_decisions_map = {}
    for review_config in review_configs:
        action_name = review_config.get("action_name")
        allowed_decisions = review_config.get("allowed_decisions", [])
        allowed_decisions_map[action_name] = allowed_decisions

    # Collect decisions for all action requests
    decisions = []

    for action_request in action_requests:
        tool_name = action_request.get("name")
        tool_args = action_request.get("args", {})

        # Get allowed decisions for this action
        allowed_decisions = allowed_decisions_map.get(
            tool_name, ["approve", "reject"]
        )

        # Build permission options based on allowed decisions
        options = []
        if "approve" in allowed_decisions:
            options.append(
                PermissionOption(
                    optionId="allow-once",
                    name="Allow once",
                    kind="allow_once",
                )
            )
        if "reject" in allowed_decisions:
            options.append(
                PermissionOption(
                    optionId="reject-once",
                    name="Reject",
                    kind="reject_once",
                )
            )
        # Generate a tool call ID for this permission request
        # We need to find the corresponding tool call from the stored calls
        # For now, use a generated ID
        tool_call_id = f"perm_{uuid.uuid4().hex[:8]}"

        # Create ACP ToolCall object for the permission request
        acp_tool_call = ACPToolCall(
            toolCallId=tool_call_id,
            title=tool_name,
            rawInput=tool_args,
            status="pending",
        )

        # Send permission request to client
        response = await self._connection.requestPermission(
            RequestPermissionRequest(
                sessionId=params.sessionId,
                toolCall=acp_tool_call,
                options=options,
            )
        )

        # Convert ACP response to LangGraph decision
        outcome = response.outcome

        if isinstance(outcome, AllowedOutcome):
            option_id = outcome.optionId
            if option_id == "allow-once":
                # Check if this was actually an edit option
                selected_option = next(
                    (opt for opt in options if opt.optionId == option_id), None
                )
                if selected_option and selected_option.field_meta:
                    # This is an edit - for now, just approve
                    # TODO: Implement actual edit functionality
                    decisions.append({"type": "approve"})
                else:
                    decisions.append({"type": "approve"})
            elif option_id == "edit":
                # Edit option - for now, just approve
                # TODO: Implement actual edit functionality to collect edited args
                decisions.append({"type": "approve"})
        elif isinstance(outcome, DeniedOutcome):
            decisions.append(
                {
                    "type": "reject",
                    "message": "Action rejected by user",
                }
            )

    return decisions

async def _stream_and_handle_updates(
    self,
    params: PromptRequest,
    agent: Any,
    stream_input: dict[str, Any] | Command,
    config: dict[str, Any],
) -> list[Interrupt]:
    """Stream agent execution and handle updates, returning any interrupts.

    Args:
        params: The prompt request parameters
        agent: The agent to stream from
        stream_input: Input to pass to agent.astream (initial message or Command)
        config: Configuration with thread_id

    Returns:
        List of interrupts that occurred during streaming
    """
    interrupts = []

    async for stream_mode, data in agent.astream(
        stream_input,
        config=config,
        stream_mode=["messages", "updates"],
    ):
        if stream_mode == "messages":
            # Handle streaming message chunks (AIMessageChunk)
            message, metadata = data
            if isinstance(message, AIMessageChunk):
                await self._handle_ai_message_chunk(params, message)
        elif stream_mode == "updates":
            # Handle completed node updates
            for node_name, update in data.items():
                # Check for interrupts
                if node_name == "__interrupt__":
                    # Extract interrupts from the update
                    interrupts.extend(update)
                    continue

                # Only process model and tools nodes
                if node_name not in ("model", "tools"):
                    continue

                # Handle todos from tools node
                if node_name == "tools" and "todos" in update:
                    todos = update.get("todos", [])
                    if todos:
                        await self._handle_todo_update(params, todos)

                # Get messages from the update
                messages = update.get("messages", [])
                if not messages:
                    continue

                # Process the last message from this node
                last_message = messages[-1]

                # Handle completed AI messages from model node
                if node_name == "model" and isinstance(last_message, AIMessage):
                    # Check if this AIMessage has tool calls
                    if last_message.tool_calls:
                        await self._handle_completed_tool_calls(
                            params, last_message
                        )

                # Handle tool execution results from tools node
                elif node_name == "tools" and isinstance(last_message, ToolMessage):
                    # Look up the original tool call by ID
                    tool_call = self._tool_calls.get(last_message.tool_call_id)
                    if tool_call:
                        await self._handle_tool_message(
                            params, tool_call, last_message
                        )

    return interrupts

async def prompt(
    self,
    params: PromptRequest,
) -> PromptResponse:
    """Handle a user prompt and stream responses."""
    session_id = params.sessionId
    session = self._sessions.get(session_id)

    # Extract text from prompt content blocks
    prompt_text = ""
    for block in params.prompt:
        if hasattr(block, "text"):
            prompt_text += block.text
        elif isinstance(block, dict) and "text" in block:
            prompt_text += block["text"]

    # Stream the agent's response
    agent = session["agent"]
    thread_id = session["thread_id"]
    config = {"configurable": {"thread_id": thread_id}}

    # Start with the initial user message
    stream_input: dict[str, Any] | Command = {
        "messages": [{"role": "user", "content": prompt_text}]
    }

    # Loop until there are no more interrupts
    while True:
        # Stream and collect any interrupts
        interrupts = await self._stream_and_handle_updates(
            params, agent, stream_input, config
        )

        # If no interrupts, we're done
        if not interrupts:
            break

        # Process each interrupt and collect decisions
        all_decisions = []
        for interrupt in interrupts:
            decisions = await self._handle_interrupt(params, interrupt)
            all_decisions.extend(decisions)

        # Prepare to resume with the collected decisions
        stream_input = Command(resume={"decisions": all_decisions})

    return PromptResponse(stopReason="end_turn")

async def authenticate(self, params: Any) -> Any | None:
    """Authenticate (optional)."""
    # Authentication not required for now
    return None

async def extMethod(self, method: str, params: dict[str, Any]) -> dict[str, Any]:
    """Handle extension methods (optional)."""
    raise NotImplementedError(f"Extension method {method} not supported")

async def extNotification(self, method: str, params: dict[str, Any]) -> None:
    """Handle extension notifications (optional)."""
    pass

async def cancel(self, params: CancelNotification) -> None:
    """Cancel a running session."""
    # TODO: Implement cancellation logic
    pass

async def loadSession(
    self,
    params: LoadSessionRequest,
) -> LoadSessionResponse | None:
    """Load an existing session (optional)."""
    # Not implemented yet - would need to serialize/deserialize session state
    return None

async def setSessionMode(
    self,
    params: SetSessionModeRequest,
) -> SetSessionModeResponse | None:
    """Set session mode (optional)."""
    # Could be used to switch between different agent modes
    return None

async def setSessionModel(
    self,
    params: SetSessionModelRequest,
) -> SetSessionModelResponse | None:
    """Set session model (optional)."""
    # Not supported - model is configured at agent graph creation time
    return None

from langchain.agents.middleware import HumanInTheLoopMiddleware

@tool()
def get_weather(location: str) -> str:
    """Get the weather for a given location."""
    return f"The weather in {location} is sunny with a high of 75°F."

# Create the agent graph with default configuration
model = ChatAnthropic(
    model_name="claude-sonnet-4-5-20250929",
    max_tokens=20000,
)

agent_graph = create_deep_agent(
    model=model,
    tools=[get_weather],
    checkpointer=InMemorySaver(),
    middleware=[
        HumanInTheLoopMiddleware(
            interrupt_on={
                "get_weather": True,
            }
        )
    ],
)

# Start the ACP server
reader, writer = await stdio_streams()
AgentSideConnection(lambda conn: DeepagentsACP(conn, agent_graph), writer, reader)
await asyncio.Event().wait()