[Medium] MCP (Model Context Protocol) Integration

[marklicata]

Summary

Integrate Model Context Protocol (MCP) support to allow Amplifier to connect to external tool servers, enabling a rich ecosystem of third-party integrations.

Current State

Amplifier CLI has its own module system for tools but lacks:

• MCP server connectivity
• Standard protocol for external tools
• Community tool ecosystem compatibility
• Dynamic tool discovery from MCP servers

Background: What is MCP?

Model Context Protocol (MCP) is an open standard for connecting AI assistants to external tools and data sources. An MCP server exposes:

• Tools: Functions the AI can call
• Resources: Data/context the AI can read
• Prompts: Predefined prompt templates

Example MCP servers:

• File system access
• Database queries
• API integrations (GitHub, Slack, etc.)
• Web browsing
• Code execution sandboxes

Proposed Implementation

1. MCP Configuration

# In amplifier.yaml or profile
mcp:
  servers:
    # Stdio-based server (spawned as subprocess)
    - name: filesystem
      type: stdio
      command: "npx"
      args: ["-y", "@anthropic/mcp-server-filesystem", "/path/to/allowed"]
      
    # SSE-based server (HTTP connection)  
    - name: github
      type: sse
      url: "http://localhost:3000/mcp"
      headers:
        Authorization: "Bearer ${GITHUB_TOKEN}"
        
    # WebSocket server
    - name: database
      type: websocket
      url: "ws://localhost:8080/mcp"
      
  # Global MCP settings
  settings:
    timeout_ms: 30000
    retry_attempts: 3
    auto_reconnect: true

2. MCP Transport Types

Type	Description	Use Case
stdio	Spawn process, communicate via stdin/stdout	Local tools, CLI wrappers
sse	HTTP Server-Sent Events	Web-based servers
websocket	WebSocket connection	Real-time bidirectional

3. Tool Discovery

When Amplifier starts:

1. Connect to each configured MCP server
2. Call tools/list to discover available tools
3. Register tools with Amplifier's tool system
4. Make tools available to AI

# MCP tool discovery
async def discover_tools(server: MCPServer) -> list[Tool]:
    response = await server.request("tools/list", {})
    return [
        Tool(
            name=f"{server.name}:{tool['name']}",
            description=tool['description'],
            parameters=tool['inputSchema'],
            handler=lambda args: server.request("tools/call", {
                "name": tool['name'],
                "arguments": args
            })
        )
        for tool in response['tools']
    ]

4. Tool Naming Convention

MCP tools are namespaced:

{server_name}:{tool_name}

Examples:

• filesystem:read_file
• github:create_issue
• database:query

5. Resource Access

MCP resources provide read-only context:

mcp:
  servers:
    - name: docs
      type: stdio
      command: "mcp-docs-server"
      resources:
        - uri: "docs://api/*"
          description: "API documentation"

Resources are exposed via a special tool:

amplifier> Read the API docs for the users endpoint
[AI calls docs:read_resource with uri="docs://api/users"]

6. New Module Structure

src/amplifier_app_cli/mcp/
├── __init__.py
├── client.py          # MCP client implementation
├── transports/
│   ├── __init__.py
│   ├── stdio.py       # Stdio transport
│   ├── sse.py         # SSE transport
│   └── websocket.py   # WebSocket transport
├── server.py          # Server connection manager
├── tools.py           # Tool registration bridge
├── resources.py       # Resource access
└── config.py          # MCP configuration

7. Core Interfaces

from dataclasses import dataclass
from typing import Protocol, Any
from enum import Enum

class TransportType(Enum):
    STDIO = "stdio"
    SSE = "sse"
    WEBSOCKET = "websocket"

@dataclass
class MCPServerConfig:
    name: str
    type: TransportType
    # Stdio
    command: str | None = None
    args: list[str] | None = None
    env: dict[str, str] | None = None
    # Network
    url: str | None = None
    headers: dict[str, str] | None = None
    # Settings
    timeout_ms: int = 30000
    auto_reconnect: bool = True

@dataclass
class MCPTool:
    name: str
    description: str
    input_schema: dict[str, Any]
    server: str  # Server name for namespacing

@dataclass
class MCPResource:
    uri: str
    name: str
    description: str
    mime_type: str | None

class MCPTransport(Protocol):
    async def connect(self) -> None: ...
    async def disconnect(self) -> None: ...
    async def request(
        self, 
        method: str, 
        params: dict[str, Any]
    ) -> dict[str, Any]: ...
    async def notify(
        self, 
        method: str, 
        params: dict[str, Any]
    ) -> None: ...

class MCPClient:
    def __init__(self, config: MCPServerConfig): ...
    
    async def connect(self) -> None:
        """Connect to MCP server."""
        ...
    
    async def list_tools(self) -> list[MCPTool]:
        """Get available tools from server."""
        ...
    
    async def call_tool(
        self, 
        name: str, 
        arguments: dict[str, Any]
    ) -> Any:
        """Call a tool on the server."""
        ...
    
    async def list_resources(self) -> list[MCPResource]:
        """Get available resources."""
        ...
    
    async def read_resource(self, uri: str) -> str:
        """Read a resource by URI."""
        ...

class MCPManager:
    """Manages all MCP server connections."""
    
    def __init__(self, configs: list[MCPServerConfig]): ...
    
    async def start(self) -> None:
        """Connect to all configured servers."""
        ...
    
    async def stop(self) -> None:
        """Disconnect from all servers."""
        ...
    
    def get_all_tools(self) -> list[MCPTool]:
        """Get tools from all connected servers."""
        ...
    
    async def call_tool(
        self, 
        namespaced_name: str,  # "server:tool"
        arguments: dict[str, Any]
    ) -> Any:
        """Route tool call to correct server."""
        ...

8. Integration with Amplifier Tools

Bridge MCP tools to Amplifier's tool system:

class MCPToolBridge:
    """Bridge MCP tools to Amplifier's tool interface."""
    
    def __init__(self, mcp_manager: MCPManager): ...
    
    def create_amplifier_tools(self) -> list[AmplifierTool]:
        """Convert MCP tools to Amplifier tools."""
        tools = []
        for mcp_tool in self.mcp_manager.get_all_tools():
            tools.append(AmplifierTool(
                name=f"{mcp_tool.server}:{mcp_tool.name}",
                description=mcp_tool.description,
                parameters=mcp_tool.input_schema,
                handler=self._create_handler(mcp_tool)
            ))
        return tools
    
    def _create_handler(self, mcp_tool: MCPTool):
        async def handler(args: dict[str, Any]) -> Any:
            return await self.mcp_manager.call_tool(
                f"{mcp_tool.server}:{mcp_tool.name}",
                args
            )
        return handler

9. CLI Commands

# List configured MCP servers
amplifier mcp list

# Show server details and tools
amplifier mcp show filesystem

# Test server connection
amplifier mcp test github

# Add new server interactively
amplifier mcp add

# Remove server
amplifier mcp remove database

# Refresh tool list from servers
amplifier mcp refresh

10. Slash Commands

/mcp                    # List connected MCP servers
/mcp tools              # List all MCP tools
/mcp tools filesystem   # List tools from specific server
/mcp resources          # List available resources
/mcp status             # Connection status for all servers

11. Example MCP Servers to Support

Server	Description	Type
@anthropic/mcp-server-filesystem	File system access	stdio
@anthropic/mcp-server-github	GitHub API	stdio
@anthropic/mcp-server-postgres	PostgreSQL queries	stdio
@anthropic/mcp-server-sqlite	SQLite database	stdio
@anthropic/mcp-server-puppeteer	Web browsing	stdio
Custom servers	User-built integrations	any

12. Security Considerations

mcp:
  # Security settings
  security:
    # Require explicit approval for MCP tool calls
    require_approval: true
    
    # Allow-list for MCP tools
    allowed_tools:
      - "filesystem:read_file"
      - "filesystem:list_directory"
      # Deny by omission: filesystem:write_file not listed
    
    # Sandbox stdio servers
    sandbox:
      enabled: true
      network: false      # No network access
      filesystem: "/safe/path"  # Restricted filesystem

Acceptance Criteria

• Stdio transport connects to MCP servers via subprocess
• SSE transport connects via HTTP
• WebSocket transport connects via WS
• Tools discovered automatically on connect
• Tools registered with Amplifier tool system
• Tool calls routed to correct MCP server
• Resources can be read via MCP
• amplifier mcp list shows servers and status
• /mcp tools lists available MCP tools
• Namespaced tool names (server:tool)
• Timeout handling for slow MCP calls
• Reconnection on disconnect
• Security: tool allow-list support
• Unit tests for transports
• Integration tests with mock MCP server

Related

• Depends on: None (but benefits from Permission System for security)
• Enables: Third-party tool ecosystem

Estimated Effort

High - 3-4 weeks

Files to Create/Modify

• src/amplifier_app_cli/mcp/ (new module)
• src/amplifier_app_cli/commands/mcp.py (new CLI commands)
• src/amplifier_app_cli/commands/slash.py (MCP slash commands)
• src/amplifier_app_cli/lib/session.py (tool registration)