diff --git a/AGENTS.md b/AGENTS.md index 84edbca..69fe10e 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -10,6 +10,9 @@ 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 { @@ -17,6 +20,9 @@ classDiagram +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 { @@ -24,9 +30,28 @@ classDiagram +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 { + <> + 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: diff --git a/README.md b/README.md index 980ba36..b24a648 100644 --- a/README.md +++ b/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: diff --git a/src/agent_shell/adapters/agent_adapter_protocol.py b/src/agent_shell/adapters/agent_adapter_protocol.py index e94d441..023e135 100644 --- a/src/agent_shell/adapters/agent_adapter_protocol.py +++ b/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]: + ... diff --git a/src/agent_shell/adapters/claude_code_adapter.py b/src/agent_shell/adapters/claude_code_adapter.py index cd45744..dcafb1b 100644 --- a/src/agent_shell/adapters/claude_code_adapter.py +++ b/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} + diff --git a/src/agent_shell/adapters/copilot_cli_adapter.py b/src/agent_shell/adapters/copilot_cli_adapter.py index 1564cd7..f274ad6 100644 --- a/src/agent_shell/adapters/copilot_cli_adapter.py +++ b/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 diff --git a/src/agent_shell/adapters/opencode_adapter.py b/src/agent_shell/adapters/opencode_adapter.py index 1b68b7d..c9b2232 100644 --- a/src/agent_shell/adapters/opencode_adapter.py +++ b/src/agent_shell/adapters/opencode_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.opencode_adapter") @@ -170,3 +172,96 @@ async def cancel(self) -> None: except ProcessLookupError: pass self._active_processes.clear() + + def _config_path(self) -> Path: + return Path(os.path.expanduser("~/.config/opencode/opencode.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("mcp", {}) + + if mcp_server.type == MCPServerType.STDIO: + entry = { + "type": "local", + "command": [mcp_server.command, *mcp_server.args], + "environment": dict(mcp_server.env), + "enabled": True, + } + else: + entry = { + "type": "remote", + "url": mcp_server.url, + "headers": dict(mcp_server.headers), + "enabled": True, + } + + config["mcp"][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("mcp", {}) + if mcp_server_name not in servers: + warnings.warn( + f"MCP server '{mcp_server_name}' not found in OpenCode config", + UserWarning, + stacklevel=2, + ) + return + + del servers[mcp_server_name] + config["mcp"] = servers + self._write_config(config) + + async def list_mcp_servers(self) -> list[MCPServerSpec]: + config = self._read_config() + servers = config.get("mcp", {}) + 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: + if entry.get("type") == "local": + command_array = entry.get("command", []) + command = command_array[0] if command_array else None + args = list(command_array[1:]) + result.append(MCPServerSpec( + name=name, + type=MCPServerType.STDIO, + command=command, + args=args, + env=dict(entry.get("environment", {})), + )) + elif entry.get("type") == "remote": + 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 diff --git a/src/agent_shell/models/agent.py b/src/agent_shell/models/agent.py index 4da8933..90bcb37 100644 --- a/src/agent_shell/models/agent.py +++ b/src/agent_shell/models/agent.py @@ -1,5 +1,5 @@ from enum import StrEnum -from dataclasses import dataclass +from dataclasses import dataclass, field class AgentType(StrEnum): CLAUDE_CODE = "claude_code" @@ -8,6 +8,10 @@ class AgentType(StrEnum): COPILOT_CLI = "copilot_cli" CODEX = "codex" +class MCPServerType(StrEnum): + STDIO = "stdio" + HTTP = "http" + @dataclass class AgentResponse: response: str @@ -23,3 +27,30 @@ class StreamEvent: duration: float = 0.0 session_id: str | None = None +@dataclass +class MCPServerSpec: + name: str + type: MCPServerType + command: str | None = None + args: list[str] = field(default_factory=list) + env: dict[str, str] = field(default_factory=dict) + url: str | None = None + headers: dict[str, str] = field(default_factory=dict) + + def __post_init__(self): + if self.type == MCPServerType.STDIO: + if not self.command: + raise ValueError("STDIO MCP servers require 'command'") + if self.url: + raise ValueError("STDIO MCP servers cannot have 'url'") + if self.headers: + raise ValueError("STDIO MCP servers cannot have 'headers'") + elif self.type == MCPServerType.HTTP: + if not self.url: + raise ValueError("HTTP MCP servers require 'url'") + if self.command: + raise ValueError("HTTP MCP servers cannot have 'command'") + if self.args: + raise ValueError("HTTP MCP servers cannot have 'args'") + if self.env: + raise ValueError("HTTP MCP servers cannot have 'env'") diff --git a/src/agent_shell/shell.py b/src/agent_shell/shell.py index aec8fe0..d2da09a 100644 --- a/src/agent_shell/shell.py +++ b/src/agent_shell/shell.py @@ -2,7 +2,7 @@ from pathlib import Path from typing import AsyncIterator -from agent_shell.models.agent import AgentType, AgentResponse, StreamEvent +from agent_shell.models.agent import AgentType, AgentResponse, StreamEvent, MCPServerSpec from agent_shell.adapters.agent_adapter_protocol import AgentAdapter from agent_shell.adapters.claude_code_adapter import ClaudeCodeAdapter from agent_shell.adapters.opencode_adapter import OpenCodeAdapter @@ -88,3 +88,11 @@ async def stream( await self._adapter.cancel() raise + async def add_mcp_server(self, mcp_server: MCPServerSpec) -> None: + await self._adapter.add_mcp_server(mcp_server) + + async def remove_mcp_server(self, mcp_server_name: str) -> None: + await self._adapter.remove_mcp_server(mcp_server_name) + + async def list_mcp_servers(self) -> list[MCPServerSpec]: + return await self._adapter.list_mcp_servers() diff --git a/tests/integration/test_claude_code_mcp_integration.py b/tests/integration/test_claude_code_mcp_integration.py new file mode 100644 index 0000000..3db54b2 --- /dev/null +++ b/tests/integration/test_claude_code_mcp_integration.py @@ -0,0 +1,195 @@ +import pytest +from unittest.mock import AsyncMock, MagicMock, patch + +from agent_shell.adapters.claude_code_adapter import ClaudeCodeAdapter +from agent_shell.models.agent import MCPServerSpec, MCPServerType + + +def _make_mock_process(returncode: int = 0, stdout: bytes = b"", stderr: bytes = b""): + process = AsyncMock() + process.communicate = AsyncMock(return_value=(stdout, stderr)) + process.returncode = returncode + process.pid = 12345 + return process + + +class TestAddMcpServerStdio: + async def test_invokes_claude_mcp_add_with_user_scope(self): + # Arrange + adapter = ClaudeCodeAdapter() + spec = MCPServerSpec( + name="forgetful", + type=MCPServerType.STDIO, + command="uvx", + args=["forgetful-ai"], + ) + mock_process = _make_mock_process() + + # Act + with patch("asyncio.create_subprocess_exec", return_value=mock_process) as mock_exec: + await adapter.add_mcp_server(spec) + + # Assert + cmd_args = mock_exec.call_args[0] + assert cmd_args[0] == "claude" + assert cmd_args[1] == "mcp" + assert cmd_args[2] == "add" + assert "--scope" in cmd_args + assert cmd_args[cmd_args.index("--scope") + 1] == "user" + assert "--transport" in cmd_args + assert cmd_args[cmd_args.index("--transport") + 1] == "stdio" + assert "forgetful" in cmd_args + assert "--" in cmd_args + sep_idx = cmd_args.index("--") + assert cmd_args[sep_idx + 1] == "uvx" + assert cmd_args[sep_idx + 2] == "forgetful-ai" + + async def test_passes_env_vars_with_e_flag(self): + # Arrange + adapter = ClaudeCodeAdapter() + spec = MCPServerSpec( + name="forgetful", + type=MCPServerType.STDIO, + command="uvx", + args=["forgetful-ai"], + env={"FORGETFUL_API_KEY": "secret", "FORGETFUL_URL": "http://x"}, + ) + mock_process = _make_mock_process() + + # Act + with patch("asyncio.create_subprocess_exec", return_value=mock_process) as mock_exec: + await adapter.add_mcp_server(spec) + + # Assert + cmd_args = mock_exec.call_args[0] + e_indices = [i for i, v in enumerate(cmd_args) if v == "-e"] + assert len(e_indices) == 2 + env_values = {cmd_args[i + 1] for i in e_indices} + assert "FORGETFUL_API_KEY=secret" in env_values + assert "FORGETFUL_URL=http://x" in env_values + + async def test_raises_on_subprocess_failure(self): + # Arrange + adapter = ClaudeCodeAdapter() + spec = MCPServerSpec(name="x", type=MCPServerType.STDIO, command="uvx") + mock_process = _make_mock_process(returncode=1, stderr=b"already exists") + + # Act / Assert + with patch("asyncio.create_subprocess_exec", return_value=mock_process): + with pytest.raises(RuntimeError, match="already exists"): + await adapter.add_mcp_server(spec) + + +class TestAddMcpServerHttp: + async def test_invokes_claude_mcp_add_with_http_transport(self): + # Arrange + adapter = ClaudeCodeAdapter() + spec = MCPServerSpec( + name="remote", + type=MCPServerType.HTTP, + url="https://example.com/mcp", + ) + mock_process = _make_mock_process() + + # Act + with patch("asyncio.create_subprocess_exec", return_value=mock_process) as mock_exec: + await adapter.add_mcp_server(spec) + + # Assert + cmd_args = mock_exec.call_args[0] + assert "--transport" in cmd_args + assert cmd_args[cmd_args.index("--transport") + 1] == "http" + assert "remote" in cmd_args + assert "https://example.com/mcp" in cmd_args + assert "--" not in cmd_args # no command separator for HTTP + + async def test_passes_headers_with_H_flag(self): + # Arrange + adapter = ClaudeCodeAdapter() + spec = MCPServerSpec( + name="remote", + type=MCPServerType.HTTP, + url="https://example.com/mcp", + headers={"Authorization": "Bearer abc", "X-Trace": "1"}, + ) + mock_process = _make_mock_process() + + # Act + with patch("asyncio.create_subprocess_exec", return_value=mock_process) as mock_exec: + await adapter.add_mcp_server(spec) + + # Assert + cmd_args = mock_exec.call_args[0] + h_indices = [i for i, v in enumerate(cmd_args) if v == "--header"] + assert len(h_indices) == 2 + header_values = {cmd_args[i + 1] for i in h_indices} + assert "Authorization: Bearer abc" in header_values + assert "X-Trace: 1" in header_values + + +class TestRemoveMcpServer: + async def test_invokes_claude_mcp_remove_with_user_scope(self): + # Arrange + adapter = ClaudeCodeAdapter() + mock_process = _make_mock_process() + + # Act + with patch("asyncio.create_subprocess_exec", return_value=mock_process) as mock_exec: + await adapter.remove_mcp_server("forgetful") + + # Assert + cmd_args = mock_exec.call_args[0] + assert cmd_args[0] == "claude" + assert cmd_args[1] == "mcp" + assert cmd_args[2] == "remove" + assert "--scope" in cmd_args + assert cmd_args[cmd_args.index("--scope") + 1] == "user" + assert "forgetful" in cmd_args + + async def test_warns_if_server_not_found(self): + # Arrange + adapter = ClaudeCodeAdapter() + mock_process = _make_mock_process( + returncode=1, stderr=b"No MCP server found with name: missing" + ) + + # Act / Assert + with patch("asyncio.create_subprocess_exec", return_value=mock_process): + with pytest.warns(UserWarning, match="missing"): + await adapter.remove_mcp_server("missing") + + +class TestListMcpServers: + async def test_raises_not_implemented(self): + # Arrange + adapter = ClaudeCodeAdapter() + + # Act / Assert + with pytest.raises(NotImplementedError): + await adapter.list_mcp_servers() + + +class TestOverwriteSemantics: + async def test_add_calls_remove_then_add_when_server_exists(self): + # Arrange + adapter = ClaudeCodeAdapter() + spec = MCPServerSpec(name="forgetful", type=MCPServerType.STDIO, command="uvx") + + # First call (remove) succeeds; second call (add) succeeds. + # add_mcp_server should pre-remove to make adds idempotent. + remove_proc = _make_mock_process(returncode=0) + add_proc = _make_mock_process(returncode=0) + + # Act + with patch( + "asyncio.create_subprocess_exec", + side_effect=[remove_proc, add_proc], + ) as mock_exec: + await adapter.add_mcp_server(spec) + + # Assert + assert mock_exec.call_count == 2 + first_call_args = mock_exec.call_args_list[0][0] + second_call_args = mock_exec.call_args_list[1][0] + assert first_call_args[2] == "remove" + assert second_call_args[2] == "add" diff --git a/tests/integration/test_copilot_cli_mcp_integration.py b/tests/integration/test_copilot_cli_mcp_integration.py new file mode 100644 index 0000000..37325fb --- /dev/null +++ b/tests/integration/test_copilot_cli_mcp_integration.py @@ -0,0 +1,262 @@ +import json +from pathlib import Path + +import pytest + +from agent_shell.adapters.copilot_cli_adapter import CopilotCLIAdapter +from agent_shell.models.agent import MCPServerSpec, MCPServerType + + +@pytest.fixture +def isolated_home(tmp_path, monkeypatch): + monkeypatch.setenv("HOME", str(tmp_path)) + return tmp_path + + +def _read_config(home: Path) -> dict: + return json.loads((home / ".copilot" / "mcp-config.json").read_text()) + + +class TestAddMcpServerStdio: + async def test_writes_local_entry_to_user_config(self, isolated_home): + # Arrange + adapter = CopilotCLIAdapter() + spec = MCPServerSpec( + name="forgetful", + type=MCPServerType.STDIO, + command="uvx", + args=["forgetful-ai"], + env={"FORGETFUL_API_KEY": "secret"}, + ) + + # Act + await adapter.add_mcp_server(spec) + + # Assert + entry = _read_config(isolated_home)["mcpServers"]["forgetful"] + assert entry["type"] == "local" + assert entry["command"] == "uvx" + assert entry["args"] == ["forgetful-ai"] + assert entry["env"] == {"FORGETFUL_API_KEY": "secret"} + + async def test_creates_config_file_and_parent_dir(self, isolated_home): + # Arrange + adapter = CopilotCLIAdapter() + spec = MCPServerSpec(name="forgetful", type=MCPServerType.STDIO, command="uvx") + + # Act + await adapter.add_mcp_server(spec) + + # Assert + config_path = isolated_home / ".copilot" / "mcp-config.json" + assert config_path.exists() + + async def test_preserves_existing_servers(self, isolated_home): + # Arrange + adapter = CopilotCLIAdapter() + await adapter.add_mcp_server( + MCPServerSpec(name="other", type=MCPServerType.STDIO, command="other-cmd") + ) + + # Act + await adapter.add_mcp_server( + MCPServerSpec(name="forgetful", type=MCPServerType.STDIO, command="uvx") + ) + + # Assert + servers = _read_config(isolated_home)["mcpServers"] + assert "other" in servers + assert "forgetful" in servers + + async def test_overwrites_existing_server_with_same_name(self, isolated_home): + # Arrange + adapter = CopilotCLIAdapter() + old = MCPServerSpec(name="x", type=MCPServerType.STDIO, command="old") + new = MCPServerSpec(name="x", type=MCPServerType.STDIO, command="new") + + # Act + await adapter.add_mcp_server(old) + await adapter.add_mcp_server(new) + + # Assert + assert _read_config(isolated_home)["mcpServers"]["x"]["command"] == "new" + + +class TestAddMcpServerHttp: + async def test_writes_http_entry(self, isolated_home): + # Arrange + adapter = CopilotCLIAdapter() + spec = MCPServerSpec( + name="remote", + type=MCPServerType.HTTP, + url="https://example.com/mcp", + headers={"Authorization": "Bearer x"}, + ) + + # Act + await adapter.add_mcp_server(spec) + + # Assert + entry = _read_config(isolated_home)["mcpServers"]["remote"] + assert entry["type"] == "http" + assert entry["url"] == "https://example.com/mcp" + assert entry["headers"] == {"Authorization": "Bearer x"} + + +class TestRemoveMcpServer: + async def test_removes_existing_entry(self, isolated_home): + # Arrange + adapter = CopilotCLIAdapter() + await adapter.add_mcp_server( + MCPServerSpec(name="forgetful", type=MCPServerType.STDIO, command="uvx") + ) + + # Act + await adapter.remove_mcp_server("forgetful") + + # Assert + assert "forgetful" not in _read_config(isolated_home).get("mcpServers", {}) + + async def test_warns_when_server_not_found(self, isolated_home): + # Arrange + adapter = CopilotCLIAdapter() + + # Act / Assert + with pytest.warns(UserWarning, match="missing"): + await adapter.remove_mcp_server("missing") + + async def test_warns_when_config_file_missing(self, isolated_home): + # Arrange + adapter = CopilotCLIAdapter() + + # Act / Assert + with pytest.warns(UserWarning, match="missing"): + await adapter.remove_mcp_server("missing") + + +class TestListMcpServers: + async def test_returns_empty_when_no_config(self, isolated_home): + # Arrange + adapter = CopilotCLIAdapter() + + # Act + servers = await adapter.list_mcp_servers() + + # Assert + assert servers == [] + + async def test_round_trips_stdio_spec(self, isolated_home): + # Arrange + adapter = CopilotCLIAdapter() + spec = MCPServerSpec( + name="forgetful", + type=MCPServerType.STDIO, + command="uvx", + args=["forgetful-ai"], + env={"K": "V"}, + ) + await adapter.add_mcp_server(spec) + + # Act + servers = await adapter.list_mcp_servers() + + # Assert + assert len(servers) == 1 + listed = servers[0] + assert listed.name == "forgetful" + assert listed.type == MCPServerType.STDIO + assert listed.command == "uvx" + assert listed.args == ["forgetful-ai"] + assert listed.env == {"K": "V"} + + async def test_round_trips_http_spec(self, isolated_home): + # Arrange + adapter = CopilotCLIAdapter() + spec = MCPServerSpec( + name="remote", + type=MCPServerType.HTTP, + url="https://example.com/mcp", + headers={"Authorization": "Bearer x"}, + ) + await adapter.add_mcp_server(spec) + + # Act + servers = await adapter.list_mcp_servers() + + # Assert + assert len(servers) == 1 + listed = servers[0] + assert listed.name == "remote" + assert listed.type == MCPServerType.HTTP + assert listed.url == "https://example.com/mcp" + assert listed.headers == {"Authorization": "Bearer x"} + + async def test_skips_malformed_stdio_entry_with_warning(self, isolated_home): + # Arrange + config_path = isolated_home / ".copilot" / "mcp-config.json" + config_path.parent.mkdir(parents=True) + config_path.write_text(json.dumps({ + "mcpServers": { + "broken": {"type": "local", "args": [], "tools": ["*"]}, + "good": { + "type": "local", + "command": "uvx", + "args": ["forgetful-ai"], + "tools": ["*"], + }, + } + })) + adapter = CopilotCLIAdapter() + + # Act + with pytest.warns(UserWarning, match="broken"): + servers = await adapter.list_mcp_servers() + + # Assert + assert [s.name for s in servers] == ["good"] + + async def test_skips_malformed_http_entry_with_warning(self, isolated_home): + # Arrange + config_path = isolated_home / ".copilot" / "mcp-config.json" + config_path.parent.mkdir(parents=True) + config_path.write_text(json.dumps({ + "mcpServers": { + "no-url": {"type": "http", "tools": ["*"]}, + "good": { + "type": "http", + "url": "https://example.com/mcp", + "tools": ["*"], + }, + } + })) + adapter = CopilotCLIAdapter() + + # Act + with pytest.warns(UserWarning, match="no-url"): + servers = await adapter.list_mcp_servers() + + # Assert + assert [s.name for s in servers] == ["good"] + + async def test_skips_non_dict_entry_with_warning(self, isolated_home): + # Arrange + config_path = isolated_home / ".copilot" / "mcp-config.json" + config_path.parent.mkdir(parents=True) + config_path.write_text(json.dumps({ + "mcpServers": { + "junk": "not-an-object", + "good": { + "type": "local", + "command": "uvx", + "tools": ["*"], + }, + } + })) + adapter = CopilotCLIAdapter() + + # Act + with pytest.warns(UserWarning, match="junk"): + servers = await adapter.list_mcp_servers() + + # Assert + assert [s.name for s in servers] == ["good"] diff --git a/tests/integration/test_opencode_mcp_integration.py b/tests/integration/test_opencode_mcp_integration.py new file mode 100644 index 0000000..145ee5f --- /dev/null +++ b/tests/integration/test_opencode_mcp_integration.py @@ -0,0 +1,266 @@ +import json +from pathlib import Path + +import pytest + +from agent_shell.adapters.opencode_adapter import OpenCodeAdapter +from agent_shell.models.agent import MCPServerSpec, MCPServerType + + +@pytest.fixture +def isolated_home(tmp_path, monkeypatch): + monkeypatch.setenv("HOME", str(tmp_path)) + return tmp_path + + +def _read_config(home: Path) -> dict: + config_path = home / ".config" / "opencode" / "opencode.json" + return json.loads(config_path.read_text()) + + +class TestAddMcpServerStdio: + async def test_writes_local_entry_to_user_config(self, isolated_home): + # Arrange + adapter = OpenCodeAdapter() + spec = MCPServerSpec( + name="forgetful", + type=MCPServerType.STDIO, + command="uvx", + args=["forgetful-ai"], + env={"FORGETFUL_API_KEY": "secret"}, + ) + + # Act + await adapter.add_mcp_server(spec) + + # Assert + config = _read_config(isolated_home) + entry = config["mcp"]["forgetful"] + assert entry["type"] == "local" + assert entry["command"] == ["uvx", "forgetful-ai"] + assert entry["environment"] == {"FORGETFUL_API_KEY": "secret"} + assert entry["enabled"] is True + + async def test_creates_config_file_and_parent_dir(self, isolated_home): + # Arrange + adapter = OpenCodeAdapter() + spec = MCPServerSpec(name="forgetful", type=MCPServerType.STDIO, command="uvx") + + # Act + await adapter.add_mcp_server(spec) + + # Assert + config_path = isolated_home / ".config" / "opencode" / "opencode.json" + assert config_path.exists() + + async def test_preserves_existing_unrelated_config(self, isolated_home): + # Arrange + config_path = isolated_home / ".config" / "opencode" / "opencode.json" + config_path.parent.mkdir(parents=True) + config_path.write_text(json.dumps({"model": "claude-opus", "theme": "dark"})) + + adapter = OpenCodeAdapter() + spec = MCPServerSpec(name="forgetful", type=MCPServerType.STDIO, command="uvx") + + # Act + await adapter.add_mcp_server(spec) + + # Assert + config = _read_config(isolated_home) + assert config["model"] == "claude-opus" + assert config["theme"] == "dark" + assert "forgetful" in config["mcp"] + + async def test_overwrites_existing_server_with_same_name(self, isolated_home): + # Arrange + adapter = OpenCodeAdapter() + old = MCPServerSpec(name="x", type=MCPServerType.STDIO, command="old-cmd") + new = MCPServerSpec(name="x", type=MCPServerType.STDIO, command="new-cmd") + + # Act + await adapter.add_mcp_server(old) + await adapter.add_mcp_server(new) + + # Assert + config = _read_config(isolated_home) + assert config["mcp"]["x"]["command"] == ["new-cmd"] + + +class TestAddMcpServerHttp: + async def test_writes_remote_entry(self, isolated_home): + # Arrange + adapter = OpenCodeAdapter() + spec = MCPServerSpec( + name="remote", + type=MCPServerType.HTTP, + url="https://example.com/mcp", + headers={"Authorization": "Bearer x"}, + ) + + # Act + await adapter.add_mcp_server(spec) + + # Assert + entry = _read_config(isolated_home)["mcp"]["remote"] + assert entry["type"] == "remote" + assert entry["url"] == "https://example.com/mcp" + assert entry["headers"] == {"Authorization": "Bearer x"} + assert entry["enabled"] is True + + +class TestRemoveMcpServer: + async def test_removes_existing_entry(self, isolated_home): + # Arrange + adapter = OpenCodeAdapter() + spec = MCPServerSpec(name="forgetful", type=MCPServerType.STDIO, command="uvx") + await adapter.add_mcp_server(spec) + + # Act + await adapter.remove_mcp_server("forgetful") + + # Assert + config = _read_config(isolated_home) + assert "forgetful" not in config.get("mcp", {}) + + async def test_warns_when_server_not_found(self, isolated_home): + # Arrange + adapter = OpenCodeAdapter() + + # Act / Assert + with pytest.warns(UserWarning, match="missing"): + await adapter.remove_mcp_server("missing") + + async def test_warns_when_config_file_missing(self, isolated_home): + # Arrange + adapter = OpenCodeAdapter() + + # Act / Assert (no config file at all) + with pytest.warns(UserWarning, match="missing"): + await adapter.remove_mcp_server("missing") + + +class TestListMcpServers: + async def test_returns_empty_when_no_config(self, isolated_home): + # Arrange + adapter = OpenCodeAdapter() + + # Act + servers = await adapter.list_mcp_servers() + + # Assert + assert servers == [] + + async def test_round_trips_stdio_spec(self, isolated_home): + # Arrange + adapter = OpenCodeAdapter() + spec = MCPServerSpec( + name="forgetful", + type=MCPServerType.STDIO, + command="uvx", + args=["forgetful-ai"], + env={"K": "V"}, + ) + await adapter.add_mcp_server(spec) + + # Act + servers = await adapter.list_mcp_servers() + + # Assert + assert len(servers) == 1 + listed = servers[0] + assert listed.name == "forgetful" + assert listed.type == MCPServerType.STDIO + assert listed.command == "uvx" + assert listed.args == ["forgetful-ai"] + assert listed.env == {"K": "V"} + + async def test_round_trips_http_spec(self, isolated_home): + # Arrange + adapter = OpenCodeAdapter() + spec = MCPServerSpec( + name="remote", + type=MCPServerType.HTTP, + url="https://example.com/mcp", + headers={"Authorization": "Bearer x"}, + ) + await adapter.add_mcp_server(spec) + + # Act + servers = await adapter.list_mcp_servers() + + # Assert + assert len(servers) == 1 + listed = servers[0] + assert listed.name == "remote" + assert listed.type == MCPServerType.HTTP + assert listed.url == "https://example.com/mcp" + assert listed.headers == {"Authorization": "Bearer x"} + + async def test_skips_malformed_stdio_entry_with_warning(self, isolated_home): + # Arrange + config_path = isolated_home / ".config" / "opencode" / "opencode.json" + config_path.parent.mkdir(parents=True) + config_path.write_text(json.dumps({ + "mcp": { + "broken": {"type": "local", "command": [], "enabled": True}, + "good": { + "type": "local", + "command": ["uvx", "forgetful-ai"], + "enabled": True, + }, + } + })) + adapter = OpenCodeAdapter() + + # Act + with pytest.warns(UserWarning, match="broken"): + servers = await adapter.list_mcp_servers() + + # Assert + assert [s.name for s in servers] == ["good"] + + async def test_skips_malformed_http_entry_with_warning(self, isolated_home): + # Arrange + config_path = isolated_home / ".config" / "opencode" / "opencode.json" + config_path.parent.mkdir(parents=True) + config_path.write_text(json.dumps({ + "mcp": { + "no-url": {"type": "remote", "enabled": True}, + "good": { + "type": "remote", + "url": "https://example.com/mcp", + "enabled": True, + }, + } + })) + adapter = OpenCodeAdapter() + + # Act + with pytest.warns(UserWarning, match="no-url"): + servers = await adapter.list_mcp_servers() + + # Assert + assert [s.name for s in servers] == ["good"] + + async def test_skips_non_dict_entry_with_warning(self, isolated_home): + # Arrange + config_path = isolated_home / ".config" / "opencode" / "opencode.json" + config_path.parent.mkdir(parents=True) + config_path.write_text(json.dumps({ + "mcp": { + "junk": "not-an-object", + "good": { + "type": "local", + "command": ["uvx"], + "enabled": True, + }, + } + })) + adapter = OpenCodeAdapter() + + # Act + with pytest.warns(UserWarning, match="junk"): + servers = await adapter.list_mcp_servers() + + # Assert + assert [s.name for s in servers] == ["good"] diff --git a/tests/unit/test_mcp_server_spec.py b/tests/unit/test_mcp_server_spec.py new file mode 100644 index 0000000..0320a8c --- /dev/null +++ b/tests/unit/test_mcp_server_spec.py @@ -0,0 +1,93 @@ +import pytest + +from agent_shell.models.agent import MCPServerSpec, MCPServerType + + +class TestStdioValidation: + def test_valid_stdio_spec_constructs(self): + # Arrange / Act + spec = MCPServerSpec( + name="forgetful", + type=MCPServerType.STDIO, + command="uvx", + args=["forgetful-ai"], + env={"FORGETFUL_API_KEY": "x"}, + ) + + # Assert + assert spec.name == "forgetful" + assert spec.command == "uvx" + + def test_stdio_without_command_raises(self): + # Arrange / Act / Assert + with pytest.raises(ValueError, match="STDIO.*command"): + MCPServerSpec(name="x", type=MCPServerType.STDIO) + + def test_stdio_with_url_raises(self): + # Arrange / Act / Assert + with pytest.raises(ValueError, match="STDIO.*url"): + MCPServerSpec( + name="x", + type=MCPServerType.STDIO, + command="uvx", + url="https://example.com", + ) + + def test_stdio_with_headers_raises(self): + # Arrange / Act / Assert + with pytest.raises(ValueError, match="STDIO.*headers"): + MCPServerSpec( + name="x", + type=MCPServerType.STDIO, + command="uvx", + headers={"Authorization": "Bearer x"}, + ) + + +class TestHttpValidation: + def test_valid_http_spec_constructs(self): + # Arrange / Act + spec = MCPServerSpec( + name="remote", + type=MCPServerType.HTTP, + url="https://example.com/mcp", + headers={"Authorization": "Bearer x"}, + ) + + # Assert + assert spec.url == "https://example.com/mcp" + + def test_http_without_url_raises(self): + # Arrange / Act / Assert + with pytest.raises(ValueError, match="HTTP.*url"): + MCPServerSpec(name="x", type=MCPServerType.HTTP) + + def test_http_with_command_raises(self): + # Arrange / Act / Assert + with pytest.raises(ValueError, match="HTTP.*command"): + MCPServerSpec( + name="x", + type=MCPServerType.HTTP, + url="https://example.com", + command="uvx", + ) + + def test_http_with_args_raises(self): + # Arrange / Act / Assert + with pytest.raises(ValueError, match="HTTP.*args"): + MCPServerSpec( + name="x", + type=MCPServerType.HTTP, + url="https://example.com", + args=["foo"], + ) + + def test_http_with_env_raises(self): + # Arrange / Act / Assert + with pytest.raises(ValueError, match="HTTP.*env"): + MCPServerSpec( + name="x", + type=MCPServerType.HTTP, + url="https://example.com", + env={"KEY": "value"}, + ) diff --git a/tests/unit/test_shell_mcp.py b/tests/unit/test_shell_mcp.py new file mode 100644 index 0000000..25f841c --- /dev/null +++ b/tests/unit/test_shell_mcp.py @@ -0,0 +1,54 @@ +from unittest.mock import AsyncMock + +import pytest + +from agent_shell.shell import AgentShell +from agent_shell.models.agent import AgentType, MCPServerSpec, MCPServerType + + +class TestAgentShellMcpPassthrough: + async def test_add_mcp_server_delegates_to_adapter(self): + # Arrange + shell = AgentShell(agent_type=AgentType.CLAUDE_CODE) + shell._adapter.add_mcp_server = AsyncMock() + spec = MCPServerSpec(name="x", type=MCPServerType.STDIO, command="uvx") + + # Act + await shell.add_mcp_server(spec) + + # Assert + shell._adapter.add_mcp_server.assert_awaited_once_with(spec) + + async def test_remove_mcp_server_delegates_to_adapter(self): + # Arrange + shell = AgentShell(agent_type=AgentType.CLAUDE_CODE) + shell._adapter.remove_mcp_server = AsyncMock() + + # Act + await shell.remove_mcp_server("forgetful") + + # Assert + shell._adapter.remove_mcp_server.assert_awaited_once_with("forgetful") + + async def test_list_mcp_servers_delegates_to_adapter(self): + # Arrange + shell = AgentShell(agent_type=AgentType.OPENCODE) + expected = [MCPServerSpec(name="x", type=MCPServerType.STDIO, command="uvx")] + shell._adapter.list_mcp_servers = AsyncMock(return_value=expected) + + # Act + result = await shell.list_mcp_servers() + + # Assert + assert result is expected + shell._adapter.list_mcp_servers.assert_awaited_once() + + async def test_add_mcp_server_propagates_exceptions(self): + # Arrange + shell = AgentShell(agent_type=AgentType.CLAUDE_CODE) + shell._adapter.add_mcp_server = AsyncMock(side_effect=RuntimeError("boom")) + spec = MCPServerSpec(name="x", type=MCPServerType.STDIO, command="uvx") + + # Act / Assert + with pytest.raises(RuntimeError, match="boom"): + await shell.add_mcp_server(spec)