#!/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
