Building MCP Clients

1MCP CLIENT ARCHITECTURE
2
3┌─────────────────────────────────────────────────────┐
4│                    AI HOST APPLICATION               │
5│  ┌─────────────────────────────────────────────────┐│
6│  │                  MCP CLIENT                      ││
7│  │  ┌──────────────┐  ┌──────────────────────────┐ ││
8│  │  │  Connection  │  │    Capability Registry   │ ││
9│  │  │   Manager    │  │  ┌─────┐ ┌─────┐ ┌─────┐ │ ││
10│  │  │              │  │  │Tools│ │Rsrcs│ │Prmts│ │ ││
11│  │  └──────┬───────┘  │  └─────┘ └─────┘ └─────┘ │ ││
12│  │         │          └──────────────────────────┘ ││
13│  │  ┌──────▼───────┐  ┌──────────────────────────┐ ││
14│  │  │  Transport   │  │      Request Router      │ ││
15│  │  │   Handler    │  │  (route calls to servers)│ ││
16│  │  └──────────────┘  └──────────────────────────┘ ││
17│  └─────────────────────────────────────────────────┘│
18│                           │                          │
19└───────────────────────────┼──────────────────────────┘
20                            │
21          ┌─────────────────┼─────────────────┐
22          │                 │                 │
23          ▼                 ▼                 ▼
24   ┌────────────┐    ┌────────────┐    ┌────────────┐
25   │ MCP Server │    │ MCP Server │    │ MCP Server │
26   │  (GitHub)  │    │ (Database) │    │  (Files)   │
27   └────────────┘    └────────────┘    └────────────┘

1import asyncio
2import json
3import subprocess
4from dataclasses import dataclass, field
5from typing import Any, Optional
6import sys
7
8@dataclass
9class Tool:
10    """Represents an MCP tool."""
11    name: str
12    description: str
13    input_schema: dict
14    server_name: str
15
16@dataclass
17class MCPServerConnection:
18    """Manages connection to a single MCP server."""
19    name: str
20    command: str
21    args: list[str]
22    process: Optional[subprocess.Popen] = None
23    tools: list[Tool] = field(default_factory=list)
24    request_id: int = 0
25
26    async def connect(self) -> None:
27        """Start the server process and initialize connection."""
28        self.process = subprocess.Popen(
29            [self.command] + self.args,
30            stdin=subprocess.PIPE,
31            stdout=subprocess.PIPE,
32            stderr=subprocess.PIPE,
33            text=True,
34            bufsize=1
35        )
36
37        # Initialize the session
38        await self._initialize()
39
40        # Discover capabilities
41        await self._discover_tools()
42
43    async def _send_request(self, method: str, params: dict = None) -> dict:
44        """Send a JSON-RPC request and wait for response."""
45        self.request_id += 1
46
47        request = {
48            "jsonrpc": "2.0",
49            "id": self.request_id,
50            "method": method,
51        }
52        if params:
53            request["params"] = params
54
55        # Send request
56        self.process.stdin.write(json.dumps(request) + "\n")
57        self.process.stdin.flush()
58
59        # Read response
60        response_line = self.process.stdout.readline()
61        response = json.loads(response_line)
62
63        if "error" in response:
64            raise Exception(f"MCP Error: {response['error']}")
65
66        return response.get("result", {})
67
68    async def _initialize(self) -> None:
69        """Perform MCP initialization handshake."""
70        result = await self._send_request("initialize", {
71            "protocolVersion": "2024-11-05",
72            "capabilities": {
73                "roots": {"listChanged": True},
74            },
75            "clientInfo": {
76                "name": "PythonMCPClient",
77                "version": "1.0.0"
78            }
79        })
80
81        # Send initialized notification
82        notification = {
83            "jsonrpc": "2.0",
84            "method": "initialized"
85        }
86        self.process.stdin.write(json.dumps(notification) + "\n")
87        self.process.stdin.flush()
88
89        print(f"Connected to server: {result.get('serverInfo', {}).get('name')}")
90
91    async def _discover_tools(self) -> None:
92        """Discover available tools from the server."""
93        result = await self._send_request("tools/list")
94
95        self.tools = [
96            Tool(
97                name=tool["name"],
98                description=tool.get("description", ""),
99                input_schema=tool.get("inputSchema", {}),
100                server_name=self.name
101            )
102            for tool in result.get("tools", [])
103        ]
104
105        print(f"Discovered {len(self.tools)} tools from {self.name}")
106
107    async def call_tool(self, name: str, arguments: dict) -> Any:
108        """Call a tool on this server."""
109        result = await self._send_request("tools/call", {
110            "name": name,
111            "arguments": arguments
112        })
113        return result
114
115    async def disconnect(self) -> None:
116        """Cleanly disconnect from the server."""
117        if self.process:
118            try:
119                await self._send_request("shutdown")
120            except:
121                pass
122            self.process.terminate()
123            self.process.wait()
124
125
126class MCPClient:
127    """Main MCP client managing multiple servers."""
128
129    def __init__(self):
130        self.servers: dict[str, MCPServerConnection] = {}
131        self.all_tools: dict[str, Tool] = {}
132
133    async def add_server(
134        self,
135        name: str,
136        command: str,
137        args: list[str] = None
138    ) -> None:
139        """Add and connect to an MCP server."""
140        server = MCPServerConnection(
141            name=name,
142            command=command,
143            args=args or []
144        )
145
146        await server.connect()
147        self.servers[name] = server
148
149        # Register tools in global registry
150        for tool in server.tools:
151            self.all_tools[tool.name] = tool
152
153    def list_tools(self) -> list[Tool]:
154        """List all available tools across all servers."""
155        return list(self.all_tools.values())
156
157    async def call_tool(self, name: str, arguments: dict) -> Any:
158        """Call a tool, routing to the appropriate server."""
159        if name not in self.all_tools:
160            raise ValueError(f"Unknown tool: {name}")
161
162        tool = self.all_tools[name]
163        server = self.servers[tool.server_name]
164
165        return await server.call_tool(name, arguments)
166
167    async def disconnect_all(self) -> None:
168        """Disconnect from all servers."""
169        for server in self.servers.values():
170            await server.disconnect()
171
172
173# Example usage
174async def main():
175    client = MCPClient()
176
177    # Add servers
178    await client.add_server(
179        name="filesystem",
180        command="python",
181        args=["filesystem_server.py"]
182    )
183
184    await client.add_server(
185        name="github",
186        command="npx",
187        args=["-y", "@modelcontextprotocol/server-github"]
188    )
189
190    # List all available tools
191    print("\nAvailable tools:")
192    for tool in client.list_tools():
193        print(f"  - {tool.name}: {tool.description}")
194
195    # Call a tool
196    result = await client.call_tool("read_file", {
197        "path": "/tmp/test.txt"
198    })
199    print(f"\nTool result: {result}")
200
201    await client.disconnect_all()
202
203
204if __name__ == "__main__":
205    asyncio.run(main())

1import { Client } from "@modelcontextprotocol/sdk/client/index.js";
2import { StdioClientTransport } from "@modelcontextprotocol/sdk/client/stdio.js";
3import { spawn } from "child_process";
4
5interface Tool {
6  name: string;
7  description: string;
8  inputSchema: object;
9  serverName: string;
10}
11
12class MCPClientManager {
13  private clients: Map<string, Client> = new Map();
14  private tools: Map<string, Tool> = new Map();
15
16  async addServer(
17    name: string,
18    command: string,
19    args: string[] = []
20  ): Promise<void> {
21    // Spawn the server process
22    const serverProcess = spawn(command, args, {
23      stdio: ["pipe", "pipe", "pipe"],
24    });
25
26    // Create transport
27    const transport = new StdioClientTransport({
28      reader: serverProcess.stdout,
29      writer: serverProcess.stdin,
30    });
31
32    // Create client
33    const client = new Client(
34      {
35        name: "TypeScriptMCPClient",
36        version: "1.0.0",
37      },
38      {
39        capabilities: {},
40      }
41    );
42
43    // Connect
44    await client.connect(transport);
45    this.clients.set(name, client);
46
47    // Discover tools
48    const toolsResult = await client.listTools();
49    for (const tool of toolsResult.tools) {
50      this.tools.set(tool.name, {
51        name: tool.name,
52        description: tool.description || "",
53        inputSchema: tool.inputSchema,
54        serverName: name,
55      });
56    }
57
58    console.log(`Connected to ${name}: ${toolsResult.tools.length} tools`);
59  }
60
61  listTools(): Tool[] {
62    return Array.from(this.tools.values());
63  }
64
65  async callTool(name: string, args: Record<string, unknown>): Promise<unknown> {
66    const tool = this.tools.get(name);
67    if (!tool) {
68      throw new Error(`Unknown tool: ${name}`);
69    }
70
71    const client = this.clients.get(tool.serverName);
72    if (!client) {
73      throw new Error(`No client for server: ${tool.serverName}`);
74    }
75
76    const result = await client.callTool({
77      name,
78      arguments: args,
79    });
80
81    return result;
82  }
83
84  async disconnectAll(): Promise<void> {
85    for (const [name, client] of this.clients) {
86      await client.close();
87      console.log(`Disconnected from ${name}`);
88    }
89    this.clients.clear();
90    this.tools.clear();
91  }
92}
93
94// Example usage
95async function main() {
96  const manager = new MCPClientManager();
97
98  try {
99    // Add servers
100    await manager.addServer(
101      "filesystem",
102      "python",
103      ["filesystem_server.py"]
104    );
105
106    // List tools
107    console.log("\nAvailable tools:");
108    for (const tool of manager.listTools()) {
109      console.log(`  - ${tool.name}: ${tool.description}`);
110    }
111
112    // Call a tool
113    const result = await manager.callTool("read_file", {
114      path: "/tmp/test.txt",
115    });
116    console.log("\nResult:", result);
117
118  } finally {
119    await manager.disconnectAll();
120  }
121}
122
123main().catch(console.error);

1interface ServerConfig {
2  name: string;
3  command: string;
4  args: string[];
5  env?: Record<string, string>;
6  enabled?: boolean;
7}
8
9const MCP_SERVERS: ServerConfig[] = [
10  {
11    name: "filesystem",
12    command: "python",
13    args: ["servers/filesystem_server.py"],
14    enabled: true,
15  },
16  {
17    name: "github",
18    command: "npx",
19    args: ["-y", "@modelcontextprotocol/server-github"],
20    env: {
21      GITHUB_TOKEN: process.env.GITHUB_TOKEN || "",
22    },
23    enabled: !!process.env.GITHUB_TOKEN,
24  },
25  {
26    name: "database",
27    command: "python",
28    args: ["servers/database_server.py"],
29    env: {
30      DATABASE_URL: process.env.DATABASE_URL || "",
31    },
32    enabled: !!process.env.DATABASE_URL,
33  },
34];

1async function connectAllServers(
2  manager: MCPClientManager,
3  configs: ServerConfig[]
4): Promise<void> {
5  const enabledServers = configs.filter(c => c.enabled !== false);
6
7  console.log(`Connecting to ${enabledServers.length} servers...`);
8
9  // Connect to all servers in parallel
10  const results = await Promise.allSettled(
11    enabledServers.map(async (config) => {
12      try {
13        await manager.addServer(config.name, config.command, config.args);
14        return { name: config.name, status: "connected" };
15      } catch (error) {
16        console.error(`Failed to connect to ${config.name}:`, error);
17        return { name: config.name, status: "failed", error };
18      }
19    })
20  );
21
22  // Report results
23  const connected = results.filter(
24    r => r.status === "fulfilled" && r.value.status === "connected"
25  ).length;
26
27  console.log(`${connected}/${enabledServers.length} servers connected`);
28}

1class MCPHealthMonitor {
2  private manager: MCPClientManager;
3  private checkInterval: NodeJS.Timer | null = null;
4
5  constructor(manager: MCPClientManager) {
6    this.manager = manager;
7  }
8
9  startMonitoring(intervalMs: number = 30000): void {
10    this.checkInterval = setInterval(async () => {
11      await this.checkAllServers();
12    }, intervalMs);
13  }
14
15  stopMonitoring(): void {
16    if (this.checkInterval) {
17      clearInterval(this.checkInterval);
18      this.checkInterval = null;
19    }
20  }
21
22  private async checkAllServers(): Promise<void> {
23    for (const [name, client] of this.manager.getClients()) {
24      try {
25        // Ping server by listing tools
26        await client.listTools();
27        console.log(`[Health] ${name}: OK`);
28      } catch (error) {
29        console.error(`[Health] ${name}: FAILED`, error);
30        // Attempt reconnection
31        await this.attemptReconnect(name);
32      }
33    }
34  }
35
36  private async attemptReconnect(name: string): Promise<void> {
37    console.log(`[Health] Attempting to reconnect to ${name}...`);
38    // Implementation depends on your reconnection strategy
39  }
40}

1interface RegisteredTool {
2  name: string;
3  description: string;
4  inputSchema: object;
5  serverName: string;
6  priority: number;
7}
8
9class ToolRegistry {
10  private tools: Map<string, RegisteredTool[]> = new Map();
11
12  register(tool: RegisteredTool): void {
13    const existing = this.tools.get(tool.name) || [];
14    existing.push(tool);
15    // Sort by priority (highest first)
16    existing.sort((a, b) => b.priority - a.priority);
17    this.tools.set(tool.name, existing);
18  }
19
20  resolve(toolName: string): RegisteredTool | null {
21    const candidates = this.tools.get(toolName);
22    if (!candidates || candidates.length === 0) {
23      return null;
24    }
25    // Return highest priority tool
26    return candidates[0];
27  }
28
29  resolveWithNamespace(qualifiedName: string): RegisteredTool | null {
30    // Support namespaced tool names like "github:create_issue"
31    const [namespace, toolName] = qualifiedName.includes(":")
32      ? qualifiedName.split(":", 2)
33      : [null, qualifiedName];
34
35    const candidates = this.tools.get(toolName);
36    if (!candidates) return null;
37
38    if (namespace) {
39      return candidates.find(t => t.serverName === namespace) || null;
40    }
41
42    return candidates[0];
43  }
44
45  listAll(): RegisteredTool[] {
46    const allTools: RegisteredTool[] = [];
47    for (const [name, candidates] of this.tools) {
48      // Only include highest priority for each name
49      allTools.push(candidates[0]);
50    }
51    return allTools;
52  }
53
54  listWithConflicts(): Map<string, RegisteredTool[]> {
55    return new Map(
56      Array.from(this.tools.entries()).filter(
57        ([_, candidates]) => candidates.length > 1
58      )
59    );
60  }
61}

1class DynamicToolDiscovery {
2  private registry: ToolRegistry;
3  private manager: MCPClientManager;
4
5  constructor(manager: MCPClientManager, registry: ToolRegistry) {
6    this.manager = manager;
7    this.registry = registry;
8  }
9
10  async refreshAll(): Promise<void> {
11    // Clear registry
12    this.registry = new ToolRegistry();
13
14    // Re-discover from all servers
15    for (const [serverName, client] of this.manager.getClients()) {
16      const tools = await client.listTools();
17
18      for (const tool of tools.tools) {
19        this.registry.register({
20          name: tool.name,
21          description: tool.description || "",
22          inputSchema: tool.inputSchema,
23          serverName,
24          priority: this.getPriorityForServer(serverName),
25        });
26      }
27    }
28  }
29
30  private getPriorityForServer(serverName: string): number {
31    // Define server priorities (higher = preferred)
32    const priorities: Record<string, number> = {
33      "local-filesystem": 100,
34      "github": 90,
35      "database": 80,
36      "fallback": 10,
37    };
38    return priorities[serverName] || 50;
39  }
40
41  subscribeToChanges(callback: () => void): void {
42    // Listen for tools/list_changed notifications from servers
43    // This would require notification handling in the client
44    this.manager.onNotification("notifications/tools/list_changed", () => {
45      this.refreshAll().then(callback);
46    });
47  }
48}