Design and launch MCP agents

# Manual MCP protocol implementation
import json
import asyncio
from typing import Dict, Any

class MCPClient:
    def __init__(self, server_url: str):
        self.server_url = server_url
        self.session = None
        self.message_id = 0

    async def connect(self):
        # Handle WebSocket connection
        self.session = await create_websocket_session()
        await self._handshake()

    async def _handshake(self):
        # Implement JSON-RPC handshake
        message = self._create_message("initialize")
        response = await self._send_message(message)
        if response["error"]:
            raise Exception(response["error"])

    def _create_message(self, method: str, params: Dict = None):
        self.message_id += 1
        return {
            "jsonrpc": "2.0",
            "method": method,
            "params": params or {},
            "id": self.message_id
        }

    async def call_tool(self, tool_name: str, args: Dict):
        # Create tool call message
        message = self._create_message("tools/call", {
            "name": tool_name,
            "arguments": args
        })
        # Send and handle response
        response = await self._send_message(message)
        return self._parse_response(response)

# Plus 100+ more lines for error handling, retries, etc.

from fastapi import FastAPI
from pydantic import BaseModel
import inspect
import json

app = FastAPI()

def add(a: int, b: int) -> int:
    return a + b

# Manual tool registration
tools_registry = {}

def register_tool(func):
    sig = inspect.signature(func)
    schema = {
        "name": func.__name__,
        "description": func.__doc__,
        "inputSchema": {
            "type": "object",
            "properties": {},
            "required": []
        }
    }

    # Parse parameters and types
    for param_name, param in sig.parameters.items():
        param_type = "string"
        if param.annotation == int:
            param_type = "integer"
        elif param.annotation == float:
            param_type = "number"

        schema["inputSchema"]["properties"][param_name] = {
            "type": param_type
        }
        schema["inputSchema"]["required"].append(param_name)

    tools_registry[func.__name__] = {
        "schema": schema,
        "func": func
    }

register_tool(add)

@app.post("/mcp")
async def handle_mcp(request: dict):
    method = request.get("method")

    if method == "tools/list":
        return {
            "tools": [t["schema"] for t in tools_registry.values()]
        }

    elif method == "tools/call":
        tool_name = request["params"]["name"]
        args = request["params"]["arguments"]

        if tool_name in tools_registry:
            func = tools_registry[tool_name]["func"]
            result = func(**args)
            return {"content": [{"type": "text", "text": str(result)}]}

    return {"error": "Method not found"}

import subprocess
import asyncio
import json
from typing import List, Dict

class StdioMCPClient:
    def __init__(self, command: List[str]):
        self.command = command
        self.process = None

    async def start(self):
        self.process = await asyncio.create_subprocess_exec(
            *self.command,
            stdin=subprocess.PIPE,
            stdout=subprocess.PIPE,
            stderr=subprocess.PIPE
        )

    async def send_message(self, message: dict):
        # Format JSON-RPC message
        json_msg = json.dumps(message)
        self.process.stdin.write(json_msg.encode() + b'\n')
        await self.process.stdin.drain()

        # Read response
        response_line = await self.process.stdout.readline()
        return json.loads(response_line)

class MultiServerManager:
    def __init__(self):
        self.http_clients = []
        self.stdio_clients = []

    async def add_http_server(self, url: str):
        client = MCPClient(url)
        await client.connect()
        self.http_clients.append(client)

    async def add_stdio_server(self, command: List[str]):
        client = StdioMCPClient(command)
        await client.start()
        # Initialize handshake
        await client.send_message({
            "jsonrpc": "2.0",
            "method": "initialize",
            "params": {},
            "id": 1
        })
        self.stdio_clients.append(client)

    async def list_all_tools(self):
        all_tools = []
        # Query HTTP servers
        for client in self.http_clients:
            tools = await client.list_tools()
            all_tools.extend(tools)

        # Query stdio servers
        for client in self.stdio_clients:
            response = await client.send_message({
                "method": "tools/list"
            })
            all_tools.extend(response["tools"])

        return all_tools

# Usage - complex setup
manager = MultiServerManager()
await manager.add_http_server("http://localhost:8000")
await manager.add_stdio_server(["npx", "@playwright/mcp@latest"])
tools = await manager.list_all_tools()