数値計算ツールMCPサーバーの技術解説
目次
- プロジェクト概要
- 全体アーキテクチャ
- 主要コンポーネント
- 通信プロトコル詳細
- math_evaluator.py 実装原理
- bridge_connector.py 実装原理
- MCPプロトコルメッセージ形式
- 実行フロー解析
- 設定駆動メカニズム
1. プロジェクト概要
1.1 MCPとは
MCP（Model Context Protocol）は、サーバーが言語モデルに公開可能な呼び出し可能ツールを提供するための標準化プロトコルです。これによりモデルは外部システムと連携できます：
- データベースクエリ
- 外部API呼び出し
- 数値計算
- メール送信
- デバイス制御
1.2 プロジェクトの用途
MCP-Calculatorは、aiVoice（AI音声アシスタント）のMCP統合サンプルプロジェクトであり、以下を示しています：
- MCPサーバーを作成しツールを公開する方法
bridge_connector.pyを介してローカルMCPサービスをaiVoiceクラウドに接続する方法- 音声端末がこれらのツールを呼び出せるようにする方法
1.3 プロジェクトファイル構造
mcp-math-calculator/
├── math_evaluator.py # MCPサーバー実装（計算ツール）
├── bridge_connector.py # WebSocket <-> stdioブリッジプログラム
├── mcp_settings.json # サーバー設定ファイル
├── dependencies.txt # 依存関係リスト
└── README.md # プロジェクト説明
2. 全体アーキテクチャ
2.1 システムアーキテクチャ図
┌─────────────────────────────────────────────────────────────────────────────┐
│ aiVoice クラウド │
│ ┌─────────────────────────────────────────────────────────────────────┐ │
│ │ MCPクライアント (WebSocket) │ │
│ │ 責務：ツール一覧照会、ツール呼び出しリクエスト、レスポンス返却 │ │
│ └─────────────────────────────────────────────────────────────────────┘ │
│ ▲ │
│ │ WebSocket (wss://api.aivoice.me/) │
│ │ │
└────────────────────────────────────┼────────────────────────────────────┘
│
┌────────────────────────────────────┼────────────────────────────────────┐
│ ローカルマシン │ │
│ ▼ │
│ ┌─────────────────────────────────────────────────────────────────────┐ │
│ │ bridge_connector.py │ │
│ │ │ │
│ │ 責務： │ │
│ │ 1. aiVoiceクラウドWebSocket接続 │ │
│ │ 2. math_evaluator.py子プロセス起動と管理 │ │
│ │ 3. WebSocketとstdio間でメッセージ転送 │ │
│ └─────────────────────────────────────────────────────────────────────┘ │
│ ▲ │
│ │ stdio（標準入出力） │
│ │ │
│ ┌─────────────────────────────────────────────────────────────────────┐ │
│ │ math_evaluator.py │ │
│ │ (MCPサーバー / FastMCP) │ │
│ │ │ │
│ │ 公開ツール：calculate(expression) -> dict │
│ └─────────────────────────────────────────────────────────────────────┘ │
│ │
└─────────────────────────────────────────────────────────────────────────┘
2.2 データフロー
音声端末 aiVoice クラウド bridge_connector.py
│ │ │
│ "1004に328を掛けた結果は？" │ │
│ ──────────────────────────────────► │ │
│ │ JSON-RPC: tools/call │
│ │ ─────────────────────────────────► │
│ │ │
│ │ JSON-RPC over WebSocket │
│ │ ◄─────────────────────────────────► │
│ │ │
│ │ JSON-RPC over stdio │
│ │ ─────────────────────────────────► │
│ │ │
│ │ │ ▼
│ │ math_evaluator.py
│ │ │処理
│ │ │▼
│ │ ◄───────────────────────────────── │
│ "計算結果は329312です" │ JSON-RPC レスポンス │
│ ◄───────────────────────────────── │ │
3. 主要コンポーネント
3.1 math_evaluator.py - MCPサーバー
役割：FastMCPフレームワークを使用してMCPサーバーを作成し、数学計算ツールを公開
主要コード：
from fastmcp import FastMCP
import math
import random
mcp_server = FastMCP("MathCalculator") # MathCalculatorという名前のMCPサーバーを作成
@mcp_server.tool() # デコレータ：ツールを公開
def calculate(expression: str) -> dict:
"""数値計算には常にこのツールを使用してください..."""
result = eval(expression, {"__builtins__": None}, {"math": math, "random": random})
return {"status": "success", "value": result}
mcp_server.run(transport="stdio") # stdio転送モードで実行
FastMCPフレームワークの責務：
- MCPプロトコルのメッセージ解析を自動処理
- ツール登録テーブルを管理
initialize、tools/list、tools/callなどのメソッド処理- stdio経由でリクエストを受信し、レスポンスを返信
3.2 bridge_connector.py - ブリッジプロキシ
役割：WebSocket接続とローカルstdioプロセスを接続
主要責務：
- クラウド接続：aiVoiceにWebSocketクライアントとして接続
- 子プロセス起動：
math_evaluator.pyを子プロセスとして起動 - 双方向転送：
- WebSocket → 子プロセスstdin
- 子プロセスstdout → WebSocket
- 子プロセスstderr → 端末出力
3.3 mcp_settings.json - 設定センター
役割：複数のMCPサーバー設定を定義
{
"mcpServers": {
"local-math-calculator": { // サーバー名
"type": "stdio", // 転送タイプ：stdio/sse/http
"command": "python", // 起動コマンド
"args": ["-m", "math_evaluator"] // コマンド引数
},
"remote-api-server": { // 別のサーバー
"type": "sse",
"url": "https://api.example.com/mcp",
"disabled": true // 無効化
}
}
}
4. 通信プロトコル詳細
4.1 MCPプロトコル概要
MCPはJSON-RPC 2.0プロトコルをベースとしており、3つの核心メソッドがあります：
| メソッド | 方向 | 説明 |
|---|---|---|
initialize |
クライアント → サーバー | 接続確立、能力交換 |
tools/list |
クライアント → サーバー | ツール一覧取得 |
tools/call |
クライアント → サーバー | 指定ツール呼び出し |
4.2 JSON-RPCメッセージ形式
リクエスト形式：
{
"jsonrpc": "2.0",
"id": 1,
"method": "tools/call",
"params": {
"name": "calculate",
"arguments": {
"expression": "1004 * 328"
}
}
}
レスポンス形式：
{
"jsonrpc": "2.0",
"id": 1,
"result": {
"content": [
{
"type": "text",
"text": "{\"status\": true, \"value\": 329312}"
}
],
"isError": false
}
}
エラー形式：
{
"jsonrpc": "2.0",
"id": 1,
"error": {
"code": -32600,
"message": "Invalid Request"
}
}
5. math_evaluator.py 実装原理
5.1 コード逐行解析
# 1. FastMCPフレームワークのインポート
from fastmcp import FastMCP
import sys
import logging
import math
import random
logger = logging.getLogger('MathCalculator') # ロガーの作成
# 2. Windowsエンコード修正（クロスプラットフォーム互換性）
if sys.platform == 'win32':
sys.stderr.reconfigure(encoding='utf-8')
sys.stdout.reconfigure(encoding='utf-8')
# 3. MCPサーバーインスタンスの作成
mcp_server = FastMCP("MathCalculator")
# 4. ツール定義（デコレータ使用）
@mcp_server.tool()
def calculate(expression: str) -> dict:
"""
ツール説明ドキュメント（docstring）
重要：AIはこの説明に基づいてツール呼び出しのタイミングを決定
"""
# 5. ユーザー入力の数式を安全に実行
# 重要：mathとrandomの2つのモジュールのみ注入し、悪意のあるコードを防止
result = eval(expression, {"__builtins__": None}, {"math": math, "random": random})
logger.info(f"計算式: {expression}, 結果: {result}")
# 6. 構造化された結果を返す
return {"status": "success", "value": result}
# 7. サーバー起動（stdioモード）
if __name__ == "__main__":
mcp_server.run(transport="stdio")
5.2 @mcp_server.tool() デコレータ原理
ユーザーコード FastMCPフレームワーク
│ │
▼ │
@mcp_server.tool() ──────────► ツールテーブルに登録
│ {
│ "calculate": {
│ "name": "calculate",
│ "description": "...",
│ "callback": func,
│ "schema": {...}
│ }
│ }
│ │
▼ │
関数定義 フレームワークが自動的にJSON Schema生成
5.3 stdio転送モード原理
┌─────────────────────────────────────────────────────────────┐
│ math_evaluator.py │
│ │
│ stdin ◄────────────────────────────────────────────────── │
│ JSON-RPCリクエスト (bridge_connector.pyから) │
│ │
│ フレームワーク処理 │
│ │ │
│ ▼ │
│ calculate()関数呼び出し │
│ │ │
│ ▼ │
│ stdout ───────────────────────────────────────────────────► │
│ JSON-RPCレスポンス (bridge_connector.pyへ) │
└─────────────────────────────────────────────────────────────┘
stdioを使用する理由：
- シンプル：ネットワークプログラミング不要、子プロセスは標準入出力で通信
- 分離：各MCPサーバーは独立プロセスとして、互いに影響しない
- 安全：プロセスレベルの分離、障害が発生しても単一ツールにのみ影響
6. bridge_connector.py 実装原理
6.1 主要関数概観
bridge_connector.py
├── establish_connection_with_retry() # リトライ付きWebSocket接続
├── link_to_service() # サービス接続と子プロセス管理
├── forward_websocket_to_process() # WebSocket → stdin
├── forward_process_to_websocket() # stdout → WebSocket
├── redirect_process_errors() # stderr → 端末
├── construct_service_command() # 子プロセス起動コマンド構築
├── load_configuration() # 設定ファイル読み込み
└── handle_system_signals() # シグナル処理（正常終了）
6.2 メインフロー図
┌──────────────────────────────────────────────────────────────────────────┐
│ _main()関数 │
│ │
│ 1. MCP_ENDPOINT環境変数読み取り │
│ 2. 起動モード判断： │
│ ├─ コマンドライン引数あり ───► 単一スクリプト実行 (math_evaluator.py) │
│ └─ 引数なし ──────────────► mcp_settings.json読み込み、全サービス起動│
│ │
└──────────────────────────────────────────────────────────────────────────┘
│
▼
┌──────────────────────────────────────────────────────────────────────────┐
│ establish_connection_with_retry() │
│ │
│ while True: │
│ try: │
│ link_to_service(uri, service) ───────────────────────────────────┐ │
│ except: │ │
│ retry_count++ │ │ │
│ backoff = min(backoff * 2, MAX_BACKOFF) │ 指数バックオフ │ │
│ await asyncio.sleep(backoff) ─────────┴─────────────────────────┘ │
└──────────────────────────────────────────────────────────────────────────┘
│
▼
┌──────────────────────────────────────────────────────────────────────────┐
│ link_to_service() │
│ │
│ 1. await websockets.connect(uri) ─── WebSocketクライアント │
│ │
│ 2. subprocess.Popen(cmd) ─────────── 子プロセス起動 │
│ ├─ stdin = PIPE │
│ ├─ stdout = PIPE │
│ └─ stderr = PIPE │
│ │
│ 3. asyncio.gather( │
│ forward_websocket_to_process(), ─── WebSocket → stdin │
│ forward_process_to_websocket(), ─── stdout → WebSocket │
│ redirect_process_errors() ─── stderr → 端末 │
│ ) │
└──────────────────────────────────────────────────────────────────────────┘
6.3 forward_websocket_to_process() 詳細
役割：WebSocketメッセージを読み取り、子プロセスのstdinに転送
async def forward_websocket_to_process(websocket, process, service_name):
try:
while True:
# 1. WebSocketからメッセージ受信（ブロッキング）
message = await websocket.recv()
logger.debug(f"[{service_name}] << {message[:120]}...")
# 2. バイトから文字列への変換（必要な場合）
if isinstance(message, bytes):
message = message.decode('utf-8')
# 3. 子プロセスのstdinに書き込み
process.stdin.write(message + '\n') # JSON-RPCリクエスト
process.stdin.flush() # 直ちに送信
except Exception as e:
logger.error(f"[{service_name}] WebSocketからプロセスへの転送エラー: {e}")
raise
データフロー：
aiVoice クラウド bridge_connector.py math_evaluator.py
│ │ │
│ JSON-RPCリクエスト │ │
│ {"jsonrpc":"2.0",...} │ │
│ ──────────────────────────────────► │ │
│ │ │
│ │ message = await websocket.recv()
│ │ process.stdin.write(message)
│ │ ────────────────────────────► │
│ │ │
│ │ │ JSON-RPCリクエスト
│ │ │ {"jsonrpc":"2.0",...}
│ │ │ ▼
│ │ │ FastMCPフレームワーク解析
6.4 forward_process_to_websocket() 詳細
役割：子プロセスのstdoutを読み取り、WebSocketに転送
async def forward_process_to_websocket(process, websocket, service_name):
try:
while True:
# 1. 子プロセスのstdoutから1行読み取り（ブロッキング）
data = await asyncio.to_thread(process.stdout.readline)
if not data: # EOFはプロセス終了を示す
logger.info(f"[{service_name}] プロセス出力が終了しました")
break
# 2. WebSocketに送信
logger.debug(f"[{service_name}] >> {data[:120]}...")
await websocket.send(data)
except Exception as e:
logger.error(f"[{service_name}] プロセスからWebSocketへの転送エラー: {e}")
raise
データフロー：
math_evaluator.py bridge_connector.py aiVoice クラウド
│ │ │
│ JSON-RPCレスポンス │ │
│ {"jsonrpc":"2.0",...} │ │
│ ◄────────────────────────────── │ │
│ │ │
│ │ data = process.stdout.readline()
│ │ ◄────────────────────────────── │
│ │ │
│ │ await websocket.send(data) │
│ │ ──────────────────────────────► │
│ │ │
│ │ │ JSON-RPCレスポンス
6.5 再接続メカニズム詳細
指数バックオフアルゴリズム：
INITIAL_BACKOFF = 1 # 初期待機1秒
MAX_BACKOFF = 600 # 最大待機10分
# バックオフシーケンス：1s → 2s → 4s → 8s → 16s → 32s → 64s → 128s → 256s → 512s → 600s
トリガーシナリオ：
- WebSocket接続が切断された
- 子プロセスが異常終了した
- ネットワークの不安定
7. MCPプロトコルメッセージ形式
7.1 initialize - 接続確立
クライアント送信：
{
"jsonrpc": "2.0",
"id": 1,
"method": "initialize",
"params": {
"protocolVersion": "2024-11-05",
"capabilities": {},
"clientInfo": {
"name": "aiVoice-agent",
"version": "1.0.0"
}
}
}
サーバーレスポンス：
{
"jsonrpc": "2.0",
"id": 1,
"result": {
"protocolVersion": "2024-11-05",
"capabilities": {"tools": {}},
"serverInfo": {
"name": "MathCalculator",
"version": "3.2.4"
}
}
}
7.2 tools/list - ツール一覧取得
クライアント送信：
{
"jsonrpc": "2.0",
"id": 2,
"method": "tools/list",
"params": {}
}
サーバーレスポンス：
{
"jsonrpc": "2.0",
"id": 2,
"result": {
"tools": [
{
"name": "calculate",
"description": "数値計算用ツール...",
"inputSchema": {
"type": "object",
"properties": {
"expression": {
"type": "string",
"description": "評価するPython式"
}
},
"required": ["expression"]
}
}
]
}
}
7.3 tools/call - ツール呼び出し
クライアント送信：
{
"jsonrpc": "2.0",
"id": 3,
"method": "tools/call",
"params": {
"name": "calculate",
"arguments": {
"expression": "1004 * 328"
}
}
}
サーバーレスポンス：
{
"jsonrpc": "2.0",
"id": 3,
"result": {
"content": [
{
"type": "text",
"text": "{\"status\": true, \"value\": 329312}"
}
],
"isError": false
}
}
8. 実行フロー解析
8.1 完全な起動フロー
┌──────────────────────────────────────────────────────────────────────────┐
│ ユーザーによるコマンド実行 │
│ python bridge_connector.py math_evaluator.py │
└──────────────────────────────────────────────────────────────────────────┘
│
▼
┌──────────────────────────────────────────────────────────────────────────┐
│ bridge_connector.py: main()関数 │
│ │
│ 1. signal.signal(SIGINT, handle_system_signals) ─── シグナルハンドラ登録│
│ 2. endpoint_url = os.environ.get('MCP_ENDPOINT') │
│ └─ wss://api.aivoice.me/mcp/?token=xxx の取得 │
│ 3. service_arg = sys.argv[1] │
│ └─ "math_evaluator.py" │
│ 4. asyncio.run(_main()) │
└──────────────────────────────────────────────────────────────────────────┘
│
▼
┌──────────────────────────────────────────────────────────────────────────┐
│ _main() │
│ │
│ asyncio.run(establish_connection_with_retry(endpoint_url, "math_evaluator.py")) │
│ │
│ establish_connection_with_retry(uri, service): │
│ while True: │
│ try: │
│ link_to_service(uri, service) ─── 接続試行 │
│ except: │
│ retry_count++ │
│ backoff = min(backoff * 2, MAX_BACKOFF) │
│ await asyncio.sleep(backoff) ─── 待機後再試行 │
└──────────────────────────────────────────────────────────────────────────┘
│
▼
┌──────────────────────────────────────────────────────────────────────────┐
│ link_to_service(uri, service) │
│ │
│ 1. async with websockets.connect(uri) │
│ ── WebSocket接続成功 │
│ │
│ 2. cmd, env = construct_service_command("math_evaluator.py") │
│ ── [sys.executable, "math_evaluator.py"], os.environ.copy() │
│ │
│ 3. subprocess.Popen(cmd, stdin=PIPE, stdout=PIPE, stderr=PIPE) │
│ ── math_evaluator.py子プロセス起動 │
│ │
│ 4. asyncio.gather( │
│ forward_websocket_to_process(...), ─── 3つの非同期タスク起動 │
│ forward_process_to_websocket(...), ─── │
│ redirect_process_errors(...) ─── │
│ ) │
└──────────────────────────────────────────────────────────────────────────┘
8.2 ツール呼び出し完全フロー
┌─────────────────────────────────────────────────────────────────────────┐
│ ステップ1: aiVoiceクラウドがinitialize送信 │
│ │
│ WebSocket ────────────────────────────────────────────────────────────► │
│ {"jsonrpc":"2.0","method":"initialize",...} │
│ │
│ math_evaluator.py (子プロセス) │
│ │ │
│ ▼ │
│ FastMCPフレームワーク処理 │
│ │ │
│ ▼ │
│ stdout ────────────────────────────────────────────────────────────────► │
│ {"jsonrpc":"2.0","result":{"capabilities":{...}}} │
└─────────────────────────────────────────────────────────────────────────┘
┌─────────────────────────────────────────────────────────────────────────┐
│ ステップ2: aiVoiceクラウドがtools/list送信 │
│ │
│ WebSocket ────────────────────────────────────────────────────────────► │
│ {"jsonrpc":"2.0","method":"tools/list",...} │
│ │
│ math_evaluator.py │
│ │ │
│ ▼ │
│ FastMCPがツール一覧返却 │
│ │ │
│ ▼ │
│ stdout ────────────────────────────────────────────────────────────────► │
│ {"jsonrpc":"2.0","result":{"tools":[...]}} │
└─────────────────────────────────────────────────────────────────────────┘
┌─────────────────────────────────────────────────────────────────────────┐
│ ステップ3: aiVoiceクラウドがtools/call送信（計算リクエスト） │
│ │
│ WebSocket ────────────────────────────────────────────────────────────► │
│ {"jsonrpc":"2.0","method":"tools/call", │
│ "params":{"name":"calculate", │
│ "arguments":{"expression":"1004*328"}}} │
│ │
│ math_evaluator.py │
│ │ │
│ ▼ │
│ @mcp_server.tool()デコレート関数 │
│ │ │
│ ▼ │
│ result = eval("1004*328", {...}) = 329312 │
│ │ │
│ ▼ │
│ stdout ────────────────────────────────────────────────────────────────► │
│ {"jsonrpc":"2.0","result":{"content":[{"text":"..."}]}} │
└─────────────────────────────────────────────────────────────────────────┘
9. 設定駆動メカニズム
9.1 mcp_settings.json 詳細
{
"mcpServers": {
"local-math-calculator": { // サーバー名（カスタマイズ可能）
"type": "stdio", // 転送タイプ
"command": "python", // 起動コマンド
"args": ["-m", "math_evaluator"] // 引数（モジュールとして実行）
},
"remote-api-service": {
"type": "sse", // SSEタイプのリモートサービス
"url": "https://api.example.com/mcp",
"disabled": true // 無効化（起動しない）
},
"remote-http-service": {
"type": "http", // HTTPタイプ
"url": "https://api.example.com/mcp",
"disabled": true
}
}
}
9.2 設定の優先順位
コマンドライン引数 ─────────────► 直接スクリプト実行
│ │
│ (例: math_evaluator.py) │
▼ ▼
引数をパスとして使用 設定ファイルmcp_settings.jsonを使用
│
├─► MCP_CONFIG環境変数が指すファイル
│
└─► ./mcp_settings.json
9.3 マルチサービス同時起動
bridge_connector.pyが引数なしで実行された場合：
cfg = load_configuration()
servers_cfg = cfg.get("mcpServers", {})
# 無効化されていない全サービスを起動
tasks = [asyncio.create_task(establish_connection_with_retry(endpoint_url, t))
for t in enabled_services]
await asyncio.gather(*tasks) // 全サービスを同時実行
付録：重要概念クイックリファレンス
| 概念 | 説明 |
|---|---|
| **FastMCP** | MCPプロトコルに基づくPythonフレームワーク、ツール開発を簡化 |
| **stdio** | 標準入出力、親子プロセス間通信に使用 |
| **WebSocket** | 双方向リアルタイム通信プロトコル |
| **JSON-RPC 2.0** | 軽量リモートプロシージャコールプロトコル |
| **bridge_connector.py** | WebSocket ↔ stdioブリッジプログラム |
| **@mcp_server.tool()** | デコレータ、ツール関数を公開するために使用 |
参照リンク
- MCPプロトコル仕様
- FastMCPフレームワーク
- aiVoice AI音声アシスタント
- mcp-math-calculatorサンプルプロジェクト