MCPサーバ

#!/usr/bin/env python3
"""
超シンプルなMPCサーバーのサンプル
"""
import asyncio
import json
from typing import Any

class SimpleMCPServer:
    """基本的なMPCサーバー実装"""
    
    def __init__(self):
        self.tools = {
            "get_time": self.get_time,
            "echo": self.echo,
            "add": self.add
        }
    
    async def get_time(self, params: dict) -> str:
        """現在時刻を返す"""
        from datetime import datetime
        return datetime.now().strftime("%Y-%m-%d %H:%M:%S")
    
    async def echo(self, params: dict) -> str:
        """入力をそのまま返す"""
        return params.get("message", "")
    
    async def add(self, params: dict) -> int:
        """2つの数値を足す"""
        a = params.get("a", 0)
        b = params.get("b", 0)
        return a + b
    
    async def handle_request(self, request: dict) -> dict:
        """リクエストを処理"""
        method = request.get("method")
        params = request.get("params", {})
        request_id = request.get("id")
        
        # ツール一覧の取得
        if method == "tools/list":
            return {
                "jsonrpc": "2.0",
                "id": request_id,
                "result": {
                    "tools": [
                        {
                            "name": "get_time",
                            "description": "現在時刻を取得します",
                            "inputSchema": {"type": "object", "properties": {}}
                        },
                        {
                            "name": "echo",
                            "description": "メッセージをエコーします",
                            "inputSchema": {
                                "type": "object",
                                "properties": {
                                    "message": {"type": "string"}
                                }
                            }
                        },
                        {
                            "name": "add",
                            "description": "2つの数値を足します",
                            "inputSchema": {
                                "type": "object",
                                "properties": {
                                    "a": {"type": "number"},
                                    "b": {"type": "number"}
                                }
                            }
                        }
                    ]
                }
            }
        
        # ツールの実行
        elif method == "tools/call":
            tool_name = params.get("name")
            tool_params = params.get("arguments", {})
            
            if tool_name in self.tools:
                result = await self.tools[tool_name](tool_params)
                return {
                    "jsonrpc": "2.0",
                    "id": request_id,
                    "result": {
                        "content": [
                            {
                                "type": "text",
                                "text": str(result)
                            }
                        ]
                    }
                }
            else:
                return {
                    "jsonrpc": "2.0",
                    "id": request_id,
                    "error": {
                        "code": -32601,
                        "message": f"Tool not found: {tool_name}"
                    }
                }
        
        # 初期化
        elif method == "initialize":
            return {
                "jsonrpc": "2.0",
                "id": request_id,
                "result": {
                    "protocolVersion": "0.1.0",
                    "serverInfo": {
                        "name": "simple-mcp-server",
                        "version": "1.0.0"
                    },
                    "capabilities": {
                        "tools": {}
                    }
                }
            }
        
        return {
            "jsonrpc": "2.0",
            "id": request_id,
            "error": {
                "code": -32601,
                "message": f"Method not found: {method}"
            }
        }
    
    async def run(self):
        """サーバーを実行（標準入出力でJSON-RPC通信）"""
        print("MPC Server started. Waiting for requests...", file=__import__('sys').stderr)
        
        while True:
            try:
                line = input()
                if not line:
                    continue
                
                request = json.loads(line)
                response = await self.handle_request(request)
                print(json.dumps(response))
                
            except EOFError:
                break
            except Exception as e:
                print(f"Error: {e}", file=__import__('sys').stderr)

# 実行
if __name__ == "__main__":
    server = SimpleMCPServer()
    asyncio.run(server.run())

$ echo ‘{“jsonrpc”: “2.0”, “id”: 1, “method”: “initialize”, “params”: {}}’ | python3 simple_mcp_server.py
MPC Server started. Waiting for requests…
{“jsonrpc”: “2.0”, “id”: 1, “result”: {“protocolVersion”: “0.1.0”, “serverInfo”: {“name”: “simple-mcp-server”, “version”: “1.0.0”}, “capabilities”: {“tools”: {}}}}

重要なポイント：
LLM自身はコードを実行できない（セキュリティ上）
MCPサーバーが代わりに実行（ファイル操作、API呼び出し、計算など）
JSONで結果だけをやり取りすることで安全に連携
今回のサンプルは「JSON返すだけ」に見えますが、実際には：
データベース接続
外部API呼び出し
ファイルシステム操作
複雑な計算

など、LLMができない実際の処理をここで実行できる。
普通のPythonスクリプトと同じで、Ubuntu等のOSに置くだけ

📁 ファイルシステム系

filesystem – ローカルファイルの読み書き、検索
github – GitHubリポジトリの操作、PR作成、Issue管理
google-drive – Google Driveのファイル操作

🗄️ データベース系

sqlite – SQLiteデータベース操作
postgres – PostgreSQL接続・クエリ実行
mysql – MySQL/MariaDB操作

🌐 Web・API系

fetch – HTTP/APIリクエスト実行
puppeteer – ブラウザ自動操作（スクレイピング、スクリーンショット）
brave-search – Web検索機能
slack – Slackメッセージ送信、チャンネル管理

🔧 開発ツール系

memory – LLMの長期記憶（Knowledge Graph形式）
sequential-thinking – 複雑な思考を段階的に実行
everart – 画像生成（Stable Diffusion等）

📊 ビジネスツール系

google-maps – 地図検索、経路案内
sentry – エラートラッキング
aws-kb-retrieval – AWS Knowledge Base検索

必要な機能だけインストールする

インストール例
# npm経由（多くが公式提供）
npx @modelcontextprotocol/server-filesystem

# Python版
pip install mcp-server-sqlite
python -m mcp_server_sqlite