feat: unified MCP server configuration API

AGENTS.md

@@ -10,23 +10,48 @@ classDiagram
         -AgentAdapter _adapter
         +execute(cwd, prompt, ...) AgentResponse
         +stream(cwd, prompt, ...) AsyncIterator~StreamEvent~
+        +add_mcp_server(spec) None
+        +remove_mcp_server(name) None
+        +list_mcp_servers() list~MCPServerSpec~
     }
 
     class AgentAdapter {
         <<Protocol>>
         +execute(cwd, prompt, ...) AgentResponse
         +stream(cwd, prompt, ...) AsyncIterator~StreamEvent~
         +cancel() None
+        +add_mcp_server(spec) None
+        +remove_mcp_server(name) None
+        +list_mcp_servers() list~MCPServerSpec~
     }
 
     class ClaudeCodeAdapter {
         -list _active_processes
         +execute(cwd, prompt, ...) AgentResponse
         +stream(cwd, prompt, ...) AsyncIterator~StreamEvent~
         +cancel() None
+        +add_mcp_server(spec) None
+        +remove_mcp_server(name) None
+        +list_mcp_servers() NotImplementedError
         -_parse_event(event, include_thinking) list~StreamEvent~
     }
 
+    class MCPServerSpec {
+        +str name
+        +MCPServerType type
+        +str command
+        +list args
+        +dict env
+        +str url
+        +dict headers
+    }
+
+    class MCPServerType {
+        <<StrEnum>>
+        STDIO
+        HTTP
+    }
+
     class AgentResponse {
         +str response
         +float cost
@@ -53,6 +78,8 @@ classDiagram
     ClaudeCodeAdapter ..|> AgentAdapter : satisfies
     AgentShell ..> AgentResponse : returns
     AgentShell ..> StreamEvent : yields
+    AgentShell ..> MCPServerSpec : accepts/returns
+    MCPServerSpec --> MCPServerType : typed by
     ClaudeCodeAdapter ..> StreamEvent : parses NDJSON into
 ```
 
@@ -61,11 +88,40 @@ The adapter pattern uses Python's `Protocol` (structural typing) rather than ABC
 ## Supported Agents
 
 - [x] Claude Code
-- [ ] OpenCode
+- [x] OpenCode
+- [x] Copilot CLI
 - [ ] Gemini CLI
-- [ ] Copilot CLI
 - [ ] Codex
 
+## MCP Server Configuration
+
+`AgentShell` exposes a unified API for registering MCP servers across all supported agents:
+
+```python
+from agent_shell import AgentShell
+from agent_shell.models.agent import AgentType, MCPServerSpec, MCPServerType
+
+shell = AgentShell(agent_type=AgentType.CLAUDE_CODE)
+
+await shell.add_mcp_server(MCPServerSpec(
+    name="forgetful",
+    type=MCPServerType.STDIO,
+    command="uvx",
+    args=["forgetful-ai"],
+    env={"FORGETFUL_API_KEY": "..."},
+))
+```
+
+All adapters write to user-scope configuration:
+
+| Agent | Mechanism | Location |
+|-------|-----------|----------|
+| Claude Code | `claude mcp add --scope user` subprocess | `~/.claude.json` (managed by CLI) |
+| OpenCode | direct JSON file write | `~/.config/opencode/opencode.json` |
+| Copilot CLI | direct JSON file write | `~/.copilot/mcp-config.json` |
+
+Adds are idempotent (overwrite existing entries with the same name). Removes warn rather than raise when the named server is not found. `list_mcp_servers()` is not yet implemented for Claude Code.
+
 ## Test Philosophy
 
 Tests validate real functionality, not code coverage metrics. Three tiers, each with a distinct purpose:

README.md

@@ -95,6 +95,47 @@ follow_up = await shell.execute(
 
 > **Note:** OpenCode's `run` mode auto-approves all tools. The `allowed_tools` and `effort` parameters are configured via `opencode.json`, not CLI flags.
 
+## MCP Servers
+
+Register MCP servers for any supported agent through a unified API. All adapters use user-scope configuration so registrations persist across the agent's `execute`/`stream` calls.
+
+```python
+from agent_shell.shell import AgentShell
+from agent_shell.models.agent import AgentType, MCPServerSpec, MCPServerType
+
+shell = AgentShell(agent_type=AgentType.CLAUDE_CODE)
+
+# Register a stdio MCP server (e.g. forgetful) before running an eval
+await shell.add_mcp_server(MCPServerSpec(
+    name="forgetful",
+    type=MCPServerType.STDIO,
+    command="uvx",
+    args=["forgetful-ai"],
+    env={"FORGETFUL_API_KEY": "..."},
+))
+
+response = await shell.execute(
+    cwd="/path/to/project",
+    prompt="Recall any prior decisions about the auth module",
+)
+
+# Optional cleanup
+await shell.remove_mcp_server("forgetful")
+```
+
+For HTTP transport, pass `url` and `headers` instead of `command`/`args`/`env`:
+
+```python
+await shell.add_mcp_server(MCPServerSpec(
+    name="remote",
+    type=MCPServerType.HTTP,
+    url="https://example.com/mcp",
+    headers={"Authorization": "Bearer ..."},
+))
+```
+
+`add_mcp_server` overwrites an existing server with the same name. `remove_mcp_server` warns rather than raises when the named server is not found. `list_mcp_servers()` works for OpenCode and Copilot CLI; for Claude Code it currently raises `NotImplementedError`.
+
 ## Logging
 
 Agent Shell uses Python's standard `logging` module. Configure the `agent_shell` logger to capture tool calls, session IDs, costs, and errors:

src/agent_shell/adapters/agent_adapter_protocol.py

@@ -1,5 +1,5 @@
 from typing import Protocol, AsyncIterator
-from agent_shell.models.agent import AgentResponse, StreamEvent
+from agent_shell.models.agent import AgentResponse, StreamEvent, MCPServerSpec
 
 class AgentAdapter(Protocol):
     async def execute(
@@ -30,3 +30,12 @@ def stream(
 
     async def cancel(self) -> None:
         ...
+
+    async def add_mcp_server(self, mcp_server: MCPServerSpec) -> None:
+        ...
+
+    async def remove_mcp_server(self, mcp_server_name: str) -> None: 
+        ...
+
+    async def list_mcp_servers(self) -> list[MCPServerSpec]:
+        ...

src/agent_shell/adapters/claude_code_adapter.py

@@ -2,9 +2,10 @@
 import json
 import os
 import logging
+import warnings
 from typing import AsyncIterator
 
-from agent_shell.models.agent import AgentResponse, StreamEvent
+from agent_shell.models.agent import AgentResponse, StreamEvent, MCPServerSpec, MCPServerType
 from agent_shell.process_cleanup import register_process_group, unregister_process_group
 
 logger = logging.getLogger("agent_shell.claude_code_adapter")
@@ -174,6 +175,67 @@ async def cancel(self) -> None:
                 pass
         self._active_processes.clear()
 
+    async def add_mcp_server(self, mcp_server: MCPServerSpec) -> None:
+        # Pre-remove for overwrite semantics; ignore failure (server may not exist).
+        await self._run_mcp_command(
+            ["claude", "mcp", "remove", "--scope", "user", mcp_server.name],
+            raise_on_error=False,
+        )
+
+        cmd = ["claude", "mcp", "add", "--scope", "user", "--transport", mcp_server.type.value]
+
+        if mcp_server.type == MCPServerType.STDIO:
+            for key, value in mcp_server.env.items():
+                cmd.extend(["-e", f"{key}={value}"])
+            cmd.append(mcp_server.name)
+            cmd.append("--")
+            cmd.append(mcp_server.command)
+            cmd.extend(mcp_server.args)
+        else:
+            for key, value in mcp_server.headers.items():
+                cmd.extend(["--header", f"{key}: {value}"])
+            cmd.append(mcp_server.name)
+            cmd.append(mcp_server.url)
+
+        await self._run_mcp_command(cmd, raise_on_error=True)
+
+    async def remove_mcp_server(self, mcp_server_name: str) -> None:
+        cmd = ["claude", "mcp", "remove", "--scope", "user", mcp_server_name]
+        result = await self._run_mcp_command(cmd, raise_on_error=False)
+        if result["returncode"] != 0:
+            warnings.warn(
+                f"Could not remove MCP server '{mcp_server_name}': {result['stderr']}",
+                UserWarning,
+                stacklevel=2,
+            )
+
+    async def list_mcp_servers(self) -> list[MCPServerSpec]:
+        # Claude Code's `claude mcp list` outputs a flat summary, not full specs.
+        # Reconstruction would require parsing per-server `claude mcp get` output
+        # or reading ~/.claude.json directly. Tracked separately.
+        raise NotImplementedError(
+            "list_mcp_servers is not yet implemented for Claude Code"
+        )
+
+    async def _run_mcp_command(self, cmd: list[str], raise_on_error: bool) -> dict:
+        logger.debug("MCP command: %s", cmd)
+        process = await asyncio.create_subprocess_exec(
+            *cmd,
+            stdin=asyncio.subprocess.DEVNULL,
+            stdout=asyncio.subprocess.PIPE,
+            stderr=asyncio.subprocess.PIPE,
+        )
+        stdout, stderr = await process.communicate()
+        stdout_text = stdout.decode("utf-8") if stdout else ""
+        stderr_text = stderr.decode("utf-8") if stderr else ""
+
+        if process.returncode != 0 and raise_on_error:
+            raise RuntimeError(
+                f"claude mcp command failed (exit {process.returncode}): {stderr_text.strip()}"
+            )
+
+        return {"returncode": process.returncode, "stdout": stdout_text, "stderr": stderr_text}
+
 
 
 

src/agent_shell/adapters/copilot_cli_adapter.py

@@ -2,9 +2,11 @@
 import json
 import logging
 import os
+import warnings
+from pathlib import Path
 from typing import AsyncIterator
 
-from agent_shell.models.agent import AgentResponse, StreamEvent
+from agent_shell.models.agent import AgentResponse, StreamEvent, MCPServerSpec, MCPServerType
 from agent_shell.process_cleanup import register_process_group, unregister_process_group
 
 logger = logging.getLogger("agent_shell.copilot_cli_adapter")
@@ -190,3 +192,95 @@ async def cancel(self) -> None:
             except ProcessLookupError:
                 pass
         self._active_processes.clear()
+
+    def _config_path(self) -> Path:
+        return Path(os.path.expanduser("~/.copilot/mcp-config.json"))
+
+    def _read_config(self) -> dict:
+        path = self._config_path()
+        if not path.exists():
+            return {}
+        return json.loads(path.read_text())
+
+    def _write_config(self, config: dict) -> None:
+        path = self._config_path()
+        path.parent.mkdir(parents=True, exist_ok=True)
+        path.write_text(json.dumps(config, indent=2))
+
+    async def add_mcp_server(self, mcp_server: MCPServerSpec) -> None:
+        config = self._read_config()
+        config.setdefault("mcpServers", {})
+
+        if mcp_server.type == MCPServerType.STDIO:
+            entry = {
+                "type": "local",
+                "command": mcp_server.command,
+                "args": list(mcp_server.args),
+                "env": dict(mcp_server.env),
+                "tools": ["*"],
+            }
+        else:
+            entry = {
+                "type": "http",
+                "url": mcp_server.url,
+                "headers": dict(mcp_server.headers),
+                "tools": ["*"],
+            }
+
+        config["mcpServers"][mcp_server.name] = entry
+        self._write_config(config)
+
+    async def remove_mcp_server(self, mcp_server_name: str) -> None:
+        config = self._read_config()
+        servers = config.get("mcpServers", {})
+        if mcp_server_name not in servers:
+            warnings.warn(
+                f"MCP server '{mcp_server_name}' not found in Copilot CLI config",
+                UserWarning,
+                stacklevel=2,
+            )
+            return
+
+        del servers[mcp_server_name]
+        config["mcpServers"] = servers
+        self._write_config(config)
+
+    async def list_mcp_servers(self) -> list[MCPServerSpec]:
+        config = self._read_config()
+        servers = config.get("mcpServers", {})
+        result: list[MCPServerSpec] = []
+
+        for name, entry in servers.items():
+            if not isinstance(entry, dict):
+                warnings.warn(
+                    f"Skipping malformed MCP entry '{name}': expected object, got {type(entry).__name__}",
+                    UserWarning,
+                    stacklevel=2,
+                )
+                continue
+
+            try:
+                entry_type = entry.get("type")
+                if entry_type == "local":
+                    result.append(MCPServerSpec(
+                        name=name,
+                        type=MCPServerType.STDIO,
+                        command=entry.get("command"),
+                        args=list(entry.get("args", [])),
+                        env=dict(entry.get("env", {})),
+                    ))
+                elif entry_type == "http":
+                    result.append(MCPServerSpec(
+                        name=name,
+                        type=MCPServerType.HTTP,
+                        url=entry.get("url"),
+                        headers=dict(entry.get("headers", {})),
+                    ))
+            except ValueError as e:
+                warnings.warn(
+                    f"Skipping malformed MCP entry '{name}': {e}",
+                    UserWarning,
+                    stacklevel=2,
+                )
+
+        return result