"""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 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
async def main() -> None: """Main entry point for running the ACP server.""" # from deepagents_cli.agent import create_agent_with_config # from deepagents_cli.config import create_model # from deepagents_cli.tools import fetch_url, http_request, web_search # # # Create model using CLI configuration # model = create_model() # # # Setup tools - conditionally include web_search if Tavily is available # tools = [http_request, fetch_url] # if os.environ.get("TAVILY_API_KEY"): # tools.append(web_search) # # # Create CLI agent with shell access and other CLI features # # Using default assistant_id "agent" for ACP server # agent_graph, composite_backend = create_agent_with_config( # model=model, # assistant_id="agent", # tools=tools, # sandbox=None, # Local mode # sandbox_type=None, # system_prompt=None, # Use default CLI system prompt # auto_approve=False, # Require user approval for destructive operations # enable_memory=True, # Enable persistent memory # enable_skills=True, # Enable custom skills # enable_shell=True, # Enable shell access # ) # # Define default tools
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()
def cli_main() -> None: """Synchronous CLI entry point for the ACP server.""" asyncio.run(main())
if name == "main": cli_main()
