feat: Add a Minimal Agent Client Protocol (ACP) integration - works with Zed

README.md

@@ -27,6 +27,7 @@ This project comes packed with features designed for a robust and intelligent ag
     - [Task Execution](#task-execution)
     - [Using a Claude Skill (e.g., PDF Generation)](#using-a-claude-skill-eg-pdf-generation)
     - [Web Search \& Summarization (MCP Tool)](#web-search--summarization-mcp-tool)
+  - [ACP \& Zed Editor Integration](#acp--zed-editor-integration)
   - [Testing](#testing)
     - [Quick Run](#quick-run)
     - [Test Coverage](#test-coverage)
@@ -207,6 +208,34 @@ mini-agent --workspace /path/to/your/project
 
 > 📖 For more production deployment guidance, see [Production Guide](docs/PRODUCTION_GUIDE.md)
 
+## ACP & Zed Editor Integration
+
+Mini Agent supports the [Agent Communication Protocol (ACP)](https://github.com/modelcontextprotocol/protocol) for integration with code editors like Zed.
+
+**Setup in Zed Editor:**
+
+1. Install Mini Agent in development mode or as a tool
+2. Add to your Zed `settings.json`:
+
+```json
+{
+  "agent_servers": {
+    "mini-agent": {
+      "command": "/path/to/mini-agent-acp"
+    }
+  }
+}
+```
+
+The command path should be:
+- If installed via `uv tool install`: Use the output of `which mini-agent-acp`
+- If in development mode: `./mini_agent/acp/server.py`
+
+**Usage:**
+- Open Zed's agent panel with `Ctrl+Shift+P` → "Agent: Toggle Panel"
+- Select "mini-agent" from the agent dropdown
+- Start conversations with Mini Agent directly in your editor
+
 ## Usage Examples
 
 Here are a few examples of what Mini Agent can do.

mini_agent/acp/__init__.py

@@ -0,0 +1,191 @@
+"""ACP (Agent Client Protocol) bridge for Mini-Agent."""
+
+from __future__ import annotations
+
+import asyncio
+import logging
+from dataclasses import dataclass
+from pathlib import Path
+from typing import Any
+from uuid import uuid4
+
+from acp import (
+    PROTOCOL_VERSION,
+    AgentSideConnection,
+    CancelNotification,
+    InitializeRequest,
+    InitializeResponse,
+    NewSessionRequest,
+    NewSessionResponse,
+    PromptRequest,
+    PromptResponse,
+    session_notification,
+    start_tool_call,
+    stdio_streams,
+    text_block,
+    tool_content,
+    update_agent_message,
+    update_agent_thought,
+    update_tool_call,
+)
+from pydantic import field_validator
+from acp.schema import AgentCapabilities, Implementation, McpCapabilities
+
+from mini_agent.agent import Agent
+from mini_agent.cli import add_workspace_tools, initialize_base_tools
+from mini_agent.config import Config
+from mini_agent.llm import LLMClient
+from mini_agent.retry import RetryConfig as RetryConfigBase
+from mini_agent.schema import Message
+
+logger = logging.getLogger(__name__)
+
+
+try:
+    class InitializeRequestPatch(InitializeRequest):
+        @field_validator("protocolVersion", mode="before")
+        @classmethod
+        def normalize_protocol_version(cls, value: Any) -> int:
+            if isinstance(value, str):
+                try:
+                    return int(value.split(".")[0])
+                except Exception:
+                    return 1
+            if isinstance(value, (int, float)):
+                return int(value)
+            return 1
+
+    InitializeRequest = InitializeRequestPatch
+    InitializeRequest.model_rebuild(force=True)
+except Exception:  # pragma: no cover - defensive
+    logger.debug("ACP schema patch skipped")
+
+
+@dataclass
+class SessionState:
+    agent: Agent
+    cancelled: bool = False
+
+
+class MiniMaxACPAgent:
+    """Minimal ACP adapter wrapping the existing Agent runtime."""
+
+    def __init__(
+        self,
+        conn: AgentSideConnection,
+        config: Config,
+        llm: LLMClient,
+        base_tools: list,
+        system_prompt: str,
+    ):
+        self._conn = conn
+        self._config = config
+        self._llm = llm
+        self._base_tools = base_tools
+        self._system_prompt = system_prompt
+        self._sessions: dict[str, SessionState] = {}
+
+    async def initialize(self, params: InitializeRequest) -> InitializeResponse:  # noqa: ARG002
+        return InitializeResponse(
+            protocolVersion=PROTOCOL_VERSION,
+            agentCapabilities=AgentCapabilities(loadSession=False),
+            agentInfo=Implementation(name="mini-agent", title="Mini-Agent", version="0.1.0"),
+        )
+
+    async def newSession(self, params: NewSessionRequest) -> NewSessionResponse:
+        session_id = f"sess-{len(self._sessions)}-{uuid4().hex[:8]}"
+        workspace = Path(params.cwd or self._config.agent.workspace_dir).expanduser()
+        if not workspace.is_absolute():
+            workspace = workspace.resolve()
+        tools = list(self._base_tools)
+        add_workspace_tools(tools, self._config, workspace)
+        agent = Agent(llm_client=self._llm, system_prompt=self._system_prompt, tools=tools, max_steps=self._config.agent.max_steps, workspace_dir=str(workspace))
+        self._sessions[session_id] = SessionState(agent=agent)
+        return NewSessionResponse(sessionId=session_id)
+
+    async def prompt(self, params: PromptRequest) -> PromptResponse:
+        state = self._sessions.get(params.sessionId)
+        if not state:
+            return PromptResponse(stopReason="refusal")
+        state.cancelled = False
+        user_text = "\n".join(block.get("text", "") if isinstance(block, dict) else getattr(block, "text", "") for block in params.prompt)
+        state.agent.messages.append(Message(role="user", content=user_text))
+        stop_reason = await self._run_turn(state, params.sessionId)
+        return PromptResponse(stopReason=stop_reason)
+
+    async def cancel(self, params: CancelNotification) -> None:
+        state = self._sessions.get(params.sessionId)
+        if state:
+            state.cancelled = True
+
+    async def _run_turn(self, state: SessionState, session_id: str) -> str:
+        agent = state.agent
+        for _ in range(agent.max_steps):
+            if state.cancelled:
+                return "cancelled"
+            tool_schemas = [tool.to_schema() for tool in agent.tools.values()]
+            try:
+                response = await agent.llm.generate(messages=agent.messages, tools=tool_schemas)
+            except Exception as exc:
+                logger.exception("LLM error")
+                await self._send(session_id, update_agent_message(text_block(f"Error: {exc}")))
+                return "refusal"
+            if response.thinking:
+                await self._send(session_id, update_agent_thought(text_block(response.thinking)))
+            if response.content:
+                await self._send(session_id, update_agent_message(text_block(response.content)))
+            agent.messages.append(Message(role="assistant", content=response.content, thinking=response.thinking, tool_calls=response.tool_calls))
+            if not response.tool_calls:
+                return "end_turn"
+            for call in response.tool_calls:
+                name, args = call.function.name, call.function.arguments
+                # Show tool name with key arguments for better visibility
+                args_preview = ", ".join(f"{k}={repr(v)[:50]}" for k, v in list(args.items())[:2]) if isinstance(args, dict) else ""
+                label = f"🔧 {name}({args_preview})" if args_preview else f"🔧 {name}()"
+                await self._send(session_id, start_tool_call(call.id, label, kind="execute", raw_input=args))
+                tool = agent.tools.get(name)
+                if not tool:
+                    text, status = f"❌ Unknown tool: {name}", "failed"
+                else:
+                    try:
+                        result = await tool.execute(**args)
+                        status = "completed" if result.success else "failed"
+                        prefix = "✅" if result.success else "❌"
+                        text = f"{prefix} {result.content if result.success else result.error or 'Tool execution failed'}"
+                    except Exception as exc:
+                        status, text = "failed", f"❌ Tool error: {exc}"
+                await self._send(session_id, update_tool_call(call.id, status=status, content=[tool_content(text_block(text))], raw_output=text))
+                agent.messages.append(Message(role="tool", content=text, tool_call_id=call.id, name=name))
+        return "max_turn_requests"
+
+    async def _send(self, session_id: str, update: Any) -> None:
+        await self._conn.sessionUpdate(session_notification(session_id, update))
+
+
+async def run_acp_server(config: Config | None = None) -> None:
+    """Run Mini-Agent as an ACP-compatible stdio server."""
+    config = config or Config.load()
+    logging.basicConfig(level=logging.INFO, format="%(asctime)s [%(levelname)s] %(name)s: %(message)s")
+    base_tools, skill_loader = await initialize_base_tools(config)
+    prompt_path = Config.find_config_file(config.agent.system_prompt_path)
+    if prompt_path and prompt_path.exists():
+        system_prompt = prompt_path.read_text(encoding="utf-8")
+    else:
+        system_prompt = "You are a helpful AI assistant."
+    if skill_loader:
+        meta = skill_loader.get_skills_metadata_prompt()
+        if meta:
+            system_prompt = f"{system_prompt.rstrip()}\n\n{meta}"
+    rcfg = config.llm.retry
+    llm = LLMClient(api_key=config.llm.api_key, api_base=config.llm.api_base, model=config.llm.model, retry_config=RetryConfigBase(enabled=rcfg.enabled, max_retries=rcfg.max_retries, initial_delay=rcfg.initial_delay, max_delay=rcfg.max_delay, exponential_base=rcfg.exponential_base))
+    reader, writer = await stdio_streams()
+    AgentSideConnection(lambda conn: MiniMaxACPAgent(conn, config, llm, base_tools, system_prompt), writer, reader)
+    logger.info("Mini-Agent ACP server running")
+    await asyncio.Event().wait()
+
+
+def main() -> None:
+    asyncio.run(run_acp_server())
+
+
+__all__ = ["MiniMaxACPAgent", "run_acp_server", "main"]

mini_agent/acp/server.py

@@ -0,0 +1,6 @@
+"""ACP server entry point."""
+
+from mini_agent.acp import main
+
+if __name__ == "__main__":
+    main()

mini_agent/config.py

@@ -61,6 +61,16 @@ class Config(BaseModel):
     agent: AgentConfig
     tools: ToolsConfig
 
+    @classmethod
+    def load(cls) -> "Config":
+        """Load configuration from the default search path."""
+        config_path = cls.get_default_config_path()
+        if not config_path.exists():
+            raise FileNotFoundError(
+                "Configuration file not found. Run scripts/setup-config.sh or place config.yaml in mini_agent/config/."
+            )
+        return cls.from_yaml(config_path)
+
     @classmethod
     def from_yaml(cls, config_path: str | Path) -> "Config":
         """Load configuration from YAML file

mini_agent/llm/anthropic_client.py

@@ -38,8 +38,8 @@ def __init__(
         """
         super().__init__(api_key, api_base, model, retry_config)
 
-        # Initialize Anthropic client
-        self.client = anthropic.Anthropic(
+        # Initialize Anthropic async client
+        self.client = anthropic.AsyncAnthropic(
             base_url=api_base,
             api_key=api_key,
         )
@@ -75,8 +75,8 @@ async def _make_api_request(
         if tools:
             params["tools"] = self._convert_tools(tools)
 
-        # Use Anthropic SDK's messages.create
-        response = self.client.messages.create(**params)
+        # Use Anthropic SDK's async messages.create
+        response = await self.client.messages.create(**params)
         return response
 
     def _convert_tools(self, tools: list[Any]) -> list[dict[str, Any]]:

pyproject.toml

@@ -21,10 +21,12 @@ dependencies = [
     "pipx>=1.8.0",
     "anthropic>=0.39.0",
     "openai>=1.57.4",
+    "agent-client-protocol>=0.6.0",
 ]
 
 [project.scripts]
 mini-agent = "mini_agent.cli:main"
+mini-agent-acp = "mini_agent.acp.server:main"
 
 [project.optional-dependencies]
 dev = [

tests/test_acp.py

@@ -0,0 +1,89 @@
+"""Integration tests for the MiniMax ACP adapter."""
+
+from types import SimpleNamespace
+
+import pytest
+
+from mini_agent.acp import MiniMaxACPAgent
+from mini_agent.config import AgentConfig, Config, LLMConfig, ToolsConfig
+from mini_agent.schema import FunctionCall, LLMResponse, ToolCall
+from mini_agent.tools.base import Tool, ToolResult
+
+
+class DummyConn:
+    def __init__(self):
+        self.updates = []
+
+    async def sessionUpdate(self, payload):
+        self.updates.append(payload)
+
+
+class DummyLLM:
+    def __init__(self):
+        self.calls = 0
+
+    async def generate(self, messages, tools):
+        self.calls += 1
+        if self.calls == 1:
+            return LLMResponse(
+                content="",
+                thinking="calling echo",
+                tool_calls=[
+                    ToolCall(
+                        id="tool1",
+                        type="function",
+                        function=FunctionCall(name="echo", arguments={"text": "ping"}),
+                    )
+                ],
+                finish_reason="tool",
+            )
+        return LLMResponse(content="done", thinking=None, tool_calls=None, finish_reason="stop")
+
+
+class EchoTool(Tool):
+    @property
+    def name(self):
+        return "echo"
+
+    @property
+    def description(self):
+        return "Echo helper"
+
+    @property
+    def parameters(self):
+        return {"type": "object", "properties": {"text": {"type": "string"}}}
+
+    async def execute(self, text: str):
+        return ToolResult(success=True, content=f"tool:{text}")
+
+
+@pytest.fixture
+def acp_agent(tmp_path):
+    config = Config(
+        llm=LLMConfig(api_key="test-key"),
+        agent=AgentConfig(max_steps=3, workspace_dir=str(tmp_path)),
+        tools=ToolsConfig(),
+    )
+    conn = DummyConn()
+    agent = MiniMaxACPAgent(conn, config, DummyLLM(), [EchoTool()], "system")
+    return agent, conn
+
+
+@pytest.mark.asyncio
+async def test_acp_turn_executes_tool(acp_agent):
+    agent, conn = acp_agent
+    session = await agent.newSession(SimpleNamespace(cwd=None))
+    prompt = SimpleNamespace(sessionId=session.sessionId, prompt=[{"text": "hello"}])
+    response = await agent.prompt(prompt)
+    assert response.stopReason == "end_turn"
+    assert any("tool:ping" in str(update) for update in conn.updates)
+    await agent.cancel(SimpleNamespace(sessionId=session.sessionId))
+    assert agent._sessions[session.sessionId].cancelled
+
+
+@pytest.mark.asyncio
+async def test_acp_invalid_session(acp_agent):
+    agent, _ = acp_agent
+    prompt = SimpleNamespace(sessionId="missing", prompt=[{"text": "?"}])
+    response = await agent.prompt(prompt)
+    assert response.stopReason == "refusal"