MCP サーバーの説明

概要

MCP（Model Context Protocol）は、AI アシスタントと外部システムを連携させるためのプロトコルです。本システムは完全な MCP サーバーを実装しており、8 つのツール関数を提供して、自然言語による顔追跡とパンチルト台の制御に対応しています。MCP プロトコルを通じて AI アシスタントがハードウェアを直接操作でき、インテリジェントなヒューマン・コンピュータ・インタラクションを実現します。

MCP サービスのアドレス：http://localhost:8000/mcp/

MCP プロトコルの特長

プラグアンドプレイ：Claude などの AI アシスタントとシームレスに連携
️ ツール関数：検出、制御、状態管理をカバーする 8 つの専用関数
非同期処理：同時リクエストや長時間の処理に対応
状態管理：システムの状態をリアルタイムに同期
エラー処理：充実した例外処理とエラーフィードバック
型安全性：厳密なパラメーター型の検証

ツール関数一覧

検出系ツール（1 個）

1. detect_faces()

機能：現在の顔検出結果と統計情報を取得します

入力パラメーター：なし

戻り値の例：

{
  "success": true,
  "faces": [
    {
      "id": 1,
      "confidence": 0.92,
      "bbox": {"x1": 150, "y1": 100, "x2": 250, "y2": 200},
      "center": {"x": 200, "y": 150},
      "area": 10000,
      "is_tracking": true
    }
  ],
  "count": 1,
  "timestamp": "2023-12-01T15:30:45.123Z"
}

使用シーン：

「今の画面に何人映っていますか？」
「検出された顔の信頼度はどのくらいですか？」
「どの顔を追跡していますか？」

制御系ツール（3 個）

2. control_gimbal(pan: float, tilt: float)

機能：パンチルト台を指定した絶対角度の位置へ移動させます

入力パラメーター：

pan (float)：水平角度、範囲 0～270 度
tilt (float)：垂直角度、範囲 0～270 度

戻り値の例：

{
  "success": true,
  "message": "云台已移动到指定位置",
  "position": {"pan": 135.0, "tilt": 90.0},
  "execution_time": 1.23
}

使用シーン：

「パンチルト台を水平 135 度、垂直 90 度に向けて」
「pan = 135、tilt = 135」
「右上を見て」（pan = 180、tilt = 100）

3. move_gimbal_relative(pan_offset: float, tilt_offset: float)

機能：現在の位置を基準に、角度を相対的に調整します

入力パラメーター：

pan_offset (float)：水平角度のオフセット（正の値で右方向）
tilt_offset (float)：垂直角度のオフセット（正の値で上方向）

戻り値の例：

{
  "success": true,
  "message": "云台相对移动完成",
  "previous_position": {"pan": 135.0, "tilt": 90.0},
  "new_position": {"pan": 145.0, "tilt": 85.0},
  "offset_applied": {"pan": 10.0, "tilt": -5.0}
}

使用シーン：

「右に 10 度向いて」
「少し上を向いて」
「左下に位置を調整して」

4. reset_gimbal_position()

機能：パンチルト台を初期の中央位置にリセットします

入力パラメーター：なし

戻り値の例：

{
  "success": true,
  "message": "云台已重置到初始位置",
  "position": {"pan": 135.0, "tilt": 135.0},
  "reset_time": 0.85
}

使用シーン：

「パンチルト台を中央に戻して」
「パンチルト台をリセットして」
「原点に戻して」

追跡系ツール（2 個）

5. start_face_tracking()

機能：自動顔追跡を開始します

入力パラメーター：なし

戻り値の例：

{
  "success": true,
  "message": "人脸追踪已启动",
  "tracking_config": {
    "target_center_threshold": 50,
    "pid_enabled": true,
    "kalman_enabled": true,
    "prediction_enabled": true
  },
  "current_target": {
    "id": 1,
    "center": {"x": 320, "y": 240},
    "confidence": 0.92
  }
}

使用シーン：

「顔の追跡を開始して」
「自動追従を開始して」
「パンチルト台で顔を自動追跡して」

6. stop_face_tracking()

機能：自動顔追跡を停止します

入力パラメーター：なし

戻り値の例：

{
  "success": true,
  "message": "人脸追踪已停止",
  "tracking_duration": 125.67,
  "tracking_stats": {
    "total_targets": 3,
    "target_switches": 2,
    "average_confidence": 0.89
  }
}

使用シーン：

「追跡を止めて」
「自動追従をオフにして」
「手動制御モードにして」

状態系ツール（2 個）

7. get_gimbal_status()

機能：パンチルト台の現在の状態と設定情報を取得します

入力パラメーター：なし

戻り値の例：

{
  "success": true,
  "position": {"pan": 135.0, "tilt": 90.0},
  "is_connected": true,
  "connection_info": {
    "port": "/dev/ttyS1",
    "baudrate": 115200,
    "timeout": 1.0
  },
  "capabilities": {
    "angle_range": {"pan": [0, 270], "tilt": [0, 270]},
    "precision": 1.0,
    "max_speed": 60.0
  },
  "last_command": {
    "type": "absolute_move",
    "timestamp": "2023-12-01T15:29:12.456Z",
    "execution_time": 0.95
  }
}

使用シーン：

「今、パンチルト台はどこを向いていますか？」
「パンチルト台の接続状態はどうですか？」
「パンチルト台の角度範囲は？」

8. get_system_status()

機能：システム全体の稼働状態とパフォーマンス指標を取得します

入力パラメーター：なし

戻り値の例：

{
  "success": true,
  "system": {
    "status": "running",
    "version": "2.0.0",
    "uptime": "2h 15m 32s"
  },
  "services": {
    "web": "active",
    "mcp": "active",
    "detection": "active",
    "tracking": "enabled"
  },
  "hardware": {
    "camera": {
      "device": "/dev/facecamera",
      "resolution": "640x480",
      "fps": 30,
      "status": "connected"
    },
    "gimbal": {
      "port": "/dev/ttyS1",
      "status": "connected",
      "last_response": "2023-12-01T15:30:44.123Z"
    }
  },
  "performance": {
    "detection_fps": 29.5,
    "tracking_latency": 15.2,
    "memory_usage": "385MB",
    "cpu_usage": "12%",
    "response_time": {
      "detection": 2.3,
      "gimbal_control": 18.7,
      "api_average": 5.8
    }
  }
}

使用シーン：

「システムの稼働状況はどうですか？」
「パフォーマンス指標はどうですか？」
「何か異常はありますか？」

使用例

自然言語によるインタラクション

AI アシスタントを通じて、ユーザーは自然言語でシステムを操作できます。

検出関連

ユーザー: 「今の画面に何人映っていますか？」
アシスタント: detect_faces() を呼び出し
応答: 「現在 1 人の顔を検出しています。信頼度は 92% で、画面の中央付近にいます。」

ユーザー: 「顔検出の精度はどうですか？」
アシスタント: detect_faces() → get_system_status() を呼び出し
応答: 「現在の顔検出の信頼度は 92%、システムの検出フレームレートは 29.5fps で、全体的に良好です。」

パンチルト台の制御

ユーザー: 「パンチルト台を右に 20 度回して」
アシスタント: move_gimbal_relative(pan_offset=20.0, tilt_offset=0.0) を呼び出し
応答: 「パンチルト台を右に 20 度回転させ、135° から 155° に移動しました。」

ユーザー: 「左上を見て」
アシスタント: control_gimbal(pan=100.0, tilt=100.0) を呼び出し
応答: 「パンチルト台を左上に向け、水平 100°、垂直 100° に設定しました。」

ユーザー: 「パンチルト台を中央に戻して」
アシスタント: reset_gimbal_position() を呼び出し
応答: 「パンチルト台を中央位置（135°, 135°）にリセットしました。」

追跡の制御

ユーザー: 「顔の追跡を開始して」
アシスタント: start_face_tracking() を呼び出し
応答: 「顔追跡を開始しました。ID 1 のターゲットを追跡中で、中心位置は (320, 240) です。」

ユーザー: 「追跡の調子はどうですか？」
アシスタント: get_system_status() を呼び出し
応答: 「追跡システムは正常に動作しています。遅延は 15.2ms で、PID 制御とカルマンフィルタはいずれも有効です。」

ユーザー: 「自動追従を止めて」
アシスタント: stop_face_tracking() を呼び出し
応答: 「顔追跡を停止しました。今回の追跡時間は 125.67 秒で、合計 3 つのターゲットを追跡しました。」

状態の照会

ユーザー: 「システムの状態はどうですか？」
アシスタント: get_system_status() を呼び出し
応答: 「システムは正常に稼働しており、稼働時間は 2 時間 15 分です。カメラとパンチルト台の接続は正常で、検出フレームレートは 29.5fps、メモリ使用量は 385MB、CPU 使用率は 12% です。」

ユーザー: 「パンチルト台は今どこを向いていますか？」
アシスタント: get_gimbal_status() を呼び出し
応答: 「パンチルト台の現在位置は水平 135°、垂直 90° です。デバイスの接続は正常で、前回のコマンド実行時間は 0.95 秒でした。」

プログラムからの呼び出し例

Python MCP クライアント

import asyncio
import json
from mcp_client import MCPClient

class FaceTrackerMCP:
    def __init__(self, server_url="http://localhost:8000/mcp/"):
        self.client = MCPClient(server_url)

    async def detect_faces(self):
        """顔検出結果を取得"""
        try:
            result = await self.client.call_tool("detect_faces")
            return result
        except Exception as e:
            print(f"检测失败: {e}")
            return None

    async def control_gimbal(self, pan, tilt):
        """パンチルト台の位置を制御"""
        try:
            result = await self.client.call_tool("control_gimbal", {
                "pan": pan,
                "tilt": tilt
            })
            return result
        except Exception as e:
            print(f"云台控制失败: {e}")
            return None

    async def start_tracking(self):
        """追跡を開始"""
        try:
            result = await self.client.call_tool("start_face_tracking")
            return result
        except Exception as e:
            print(f"启动追踪失败: {e}")
            return None

    async def get_status(self):
        """システム状態を取得"""
        try:
            result = await self.client.call_tool("get_system_status")
            return result
        except Exception as e:
            print(f"获取状态失败: {e}")
            return None

# 使用例
async def main():
    tracker = FaceTrackerMCP()

    # MCP サーバーに接続
    await tracker.client.connect()

    try:
        # システム状態を取得
        status = await tracker.get_status()
        print(f"系统状态: {status['system']['status']}")

        # 顔を検出
        faces = await tracker.detect_faces()
        if faces and faces['success']:
            print(f"检测到 {faces['count']} 个人脸")

        # 追跡を開始
        tracking = await tracker.start_tracking()
        if tracking and tracking['success']:
            print("人脸追踪已启动")

            # しばらく待機
            await asyncio.sleep(10)

            # パンチルト台を制御
            gimbal = await tracker.control_gimbal(150, 120)
            if gimbal and gimbal['success']:
                print(f"云台已移动到 {gimbal['position']}")

    finally:
        await tracker.client.disconnect()

# サンプルを実行
if __name__ == "__main__":
    asyncio.run(main())

JavaScript/Node.js クライアント

const { MCPClient } = require('mcp-client');

class FaceTrackerMCP {
    constructor(serverUrl = 'http://localhost:8000/mcp/') {
        this.client = new MCPClient(serverUrl);
    }

    async detectFaces() {
        try {
            const result = await this.client.callTool('detect_faces');
            return result;
        } catch (error) {
            console.error('检测失败:', error);
            return null;
        }
    }

    async controlGimbal(pan, tilt) {
        try {
            const result = await this.client.callTool('control_gimbal', {
                pan: pan,
                tilt: tilt
            });
            return result;
        } catch (error) {
            console.error('云台控制失败:', error);
            return null;
        }
    }

    async moveGimbalRelative(panOffset, tiltOffset) {
        try {
            const result = await this.client.callTool('move_gimbal_relative', {
                pan_offset: panOffset,
                tilt_offset: tiltOffset
            });
            return result;
        } catch (error) {
            console.error('相对移动失败:', error);
            return null;
        }
    }

    async startTracking() {
        try {
            const result = await this.client.callTool('start_face_tracking');
            return result;
        } catch (error) {
            console.error('启动追踪失败:', error);
            return null;
        }
    }

    async stopTracking() {
        try {
            const result = await this.client.callTool('stop_face_tracking');
            return result;
        } catch (error) {
            console.error('停止追踪失败:', error);
            return null;
        }
    }

    async getSystemStatus() {
        try {
            const result = await this.client.callTool('get_system_status');
            return result;
        } catch (error) {
            console.error('获取状态失败:', error);
            return null;
        }
    }
}

// 使用例
(async () => {
    const tracker = new FaceTrackerMCP();

    try {
        await tracker.client.connect();
        console.log('已连接到MCP服务器');

        // 一連の操作を実行
        const status = await tracker.getSystemStatus();
        console.log('系统状态:', status?.system?.status);

        const faces = await tracker.detectFaces();
        console.log('检测到人脸数量:', faces?.count);

        await tracker.startTracking();
        console.log('追踪已启动');

        // パンチルト台を制御
        await tracker.moveGimbalRelative(10, -5);
        console.log('云台已调整');

        // しばらく待ってから停止
        setTimeout(async () => {
            await tracker.stopTracking();
            console.log('追踪已停止');
            await tracker.client.disconnect();
        }, 15000);

    } catch (error) {
        console.error('操作失败:', error);
    }
})();

セキュリティ制限と検証

パラメーターの検証

すべての MCP ツール関数で、厳密なパラメーター検証を実装しています。

# 角度範囲の検証
def validate_angle(angle, name):
    if not isinstance(angle, (int, float)):
        raise ValueError(f"{name} 必须是数字类型")
    if not (0 <= angle <= 270):
        raise ValueError(f"{name} 必须在0-270度范围内，当前值: {angle}")

# 例：パンチルト台制御パラメーターの検証
@mcp.tool()
async def control_gimbal(pan: float, tilt: float):
    validate_angle(pan, "pan")
    validate_angle(tilt, "tilt")
    # ... 制御ロジックを実行

同時実行の制御

パンチルト台の制御：コマンドの競合を避けるため、直列に実行します
検出機能：同時アクセスに対応しています
状態の照会：同時実行数の制限はありません
追跡の制御：開始／停止は排他的に実行します

エラー処理

共通のエラー応答形式

# 統一エラー応答形式
{
    "success": False,
    "error": "用户友好的错误描述",
    "error_type": "ANGLE_OUT_OF_RANGE",
    "error_code": "GIMBAL_001",
    "details": {
        "provided_angle": 300.0,
        "valid_range": [0, 270],
        "timestamp": "2023-12-01T15:30:45.123Z"
    }
}

よくあるエラーの種類

エラー種別	説明	対処方法
`ANGLE_OUT_OF_RANGE`	パンチルト台の角度が範囲外です	角度の値が 0～270 度の範囲内か確認してください
`GIMBAL_NOT_CONNECTED`	パンチルト台のデバイスが接続されていません	シリアルポートの接続と権限を確認してください
`TRACKING_ALREADY_ACTIVE`	追跡はすでに実行中です	先に現在の追跡を停止してください
`NO_FACES_DETECTED`	顔が検出されていません	カメラが正常に動作しているか確認してください
`SYSTEM_BUSY`	システムが別の処理を実行中です	現在の処理が完了するまでお待ちください

MCP クライアントの設定

Claude Desktop との連携

Claude Desktop の設定ファイルに MCP サーバーを追加します。設定ファイルの場所は以下のとおりです。

Windows：%APPDATA%/Claude/claude_desktop_config.json　macOS：~/Library/Application Support/Claude/claude_desktop_config.json　Linux：~/.config/Claude/claude_desktop_config.json

{
  "mcpServers": {
    "face-tracker": {
      "command": "curl",
      "args": [
        "-X", "POST",
        "http://localhost:8000/mcp/",
        "-H", "Content-Type: application/json"
      ],
      "env": {
        "MCP_SERVER_URL": "http://localhost:8000/mcp/"
      }
    }
  }
}

カスタム MCP プロキシ

より複雑な連携が必要な場合は、MCP プロキシサービスを作成できます。

# mcp_proxy.py
import asyncio
import httpx
from mcp import create_mcp_server

class FaceTrackerProxy:
    def __init__(self, base_url="http://localhost:8000/mcp/"):
        self.base_url = base_url
        self.client = httpx.AsyncClient(timeout=30.0)

    async def proxy_tool_call(self, tool_name, params=None):
        """ツール呼び出しを実際の MCP サーバーへ中継"""
        try:
            response = await self.client.post(
                self.base_url,
                json={
                    "method": "tools/call",
                    "params": {
                        "name": tool_name,
                        "arguments": params or {}
                    }
                }
            )
            return response.json()
        except Exception as e:
            return {
                "success": False,
                "error": f"代理调用失败: {str(e)}"
            }

# MCP プロキシサーバーを作成
async def create_proxy_server():
    proxy = FaceTrackerProxy()
    server = create_mcp_server("face-tracker-proxy")

    # すべてのプロキシツールを登録
    @server.tool("detect_faces")
    async def detect_faces():
        return await proxy.proxy_tool_call("detect_faces")

    @server.tool("control_gimbal")
    async def control_gimbal(pan: float, tilt: float):
        return await proxy.proxy_tool_call("control_gimbal", {
            "pan": pan, "tilt": tilt
        })

    # ... その他のツールを登録

    return server

if __name__ == "__main__":
    server = asyncio.run(create_proxy_server())
    server.run(port=8001)

監視とデバッグ

MCP サービスのログ

# MCP 関連のログを表示
tail -f logs/app.log | grep "MCP"

# ツール呼び出しのログを表示
grep "tool_call" logs/app.log

# エラーログを表示
grep "ERROR.*MCP" logs/app.log

デバッグツール

# MCP サービスの疎通を確認
curl -X POST http://localhost:8000/mcp/ \
  -H "Content-Type: application/json" \
  -d '{"method": "tools/list"}'

# 特定のツールをテスト
curl -X POST http://localhost:8000/mcp/ \
  -H "Content-Type: application/json" \
  -d '{
    "method": "tools/call",
    "params": {
      "name": "get_system_status",
      "arguments": {}
    }
  }'

# パンチルト台の制御をテスト
curl -X POST http://localhost:8000/mcp/ \
  -H "Content-Type: application/json" \
  -d '{
    "method": "tools/call",
    "params": {
      "name": "control_gimbal",
      "arguments": {"pan": 150.0, "tilt": 120.0}
    }
  }'

パフォーマンス監視

# MCP パフォーマンス監視スクリプト
import time
import asyncio
import statistics
from mcp_client import MCPClient

async def benchmark_mcp_tools():
    client = MCPClient("http://localhost:8000/mcp/")
    await client.connect()

    # 各ツールの応答時間をテスト
    tools_to_test = [
        ("detect_faces", {}),
        ("get_system_status", {}),
        ("get_gimbal_status", {}),
    ]

    results = {}

    for tool_name, params in tools_to_test:
        times = []
        for _ in range(10):
            start = time.time()
            try:
                await client.call_tool(tool_name, params)
                end = time.time()
                times.append(end - start)
            except Exception as e:
                print(f"{tool_name} 失败: {e}")

        if times:
            results[tool_name] = {
                "avg": statistics.mean(times),
                "min": min(times),
                "max": max(times),
                "count": len(times)
            }

    await client.disconnect()

    # 結果を出力
    print("MCP工具性能测试结果:")
    for tool, stats in results.items():
        print(f"{tool}: 平均{stats['avg']:.3f}s, "
              f"最小{stats['min']:.3f}s, "
              f"最大{stats['max']:.3f}s")

if __name__ == "__main__":
    asyncio.run(benchmark_mcp_tools())

拡張開発

新しいツール関数の追加

mcp_service.py に新しいツールを追加します。

@self.mcp.tool()
async def new_detection_mode(mode: str, threshold: float = 0.7):
    """検出モードを切り替え"""
    try:
        # パラメーターを検証
        if mode not in ["fast", "accurate", "balanced"]:
            return {
                "success": False,
                "error": f"无效的检测模式: {mode}",
                "valid_modes": ["fast", "accurate", "balanced"]
            }

        # 下位の API を呼び出し
        response = await self.http_client.post(
            f"{self.web_base_url}/api/detection/mode",
            json={"mode": mode, "threshold": threshold}
        )

        if response.status_code == 200:
            data = response.json()
            return {
                "success": True,
                "message": f"检测模式已切换到 {mode}",
                "config": data
            }
        else:
            return self._handle_http_error(response, "切换检测模式")

    except Exception as e:
        return {
            "success": False,
            "error": f"切换检测模式失败: {str(e)}"
        }

対応する Web API で、基盤となる機能を実装します。

# face_tracker_mcp/api/routes.py
@bp.route('/api/detection/mode', methods=['POST'])
def set_detection_mode():
    data = request.get_json()
    mode = data.get('mode')
    threshold = data.get('threshold', 0.7)

    try:
        # モード切り替えロジックを実装
        detector.set_mode(mode, threshold)
        return jsonify({
            "success": True,
            "mode": mode,
            "threshold": threshold
        })
    except Exception as e:
        return jsonify({
            "success": False,
            "error": str(e)
        }), 500

カスタムエラー処理

def create_error_response(error_type, message, details=None):
    """標準のエラー応答を作成"""
    return {
        "success": False,
        "error": message,
        "error_type": error_type,
        "timestamp": datetime.now().isoformat(),
        "details": details or {}
    }

ツール関数のデコレーター

def mcp_tool_wrapper(func):
    """MCP ツール関数用デコレーター。統一的なエラー処理とログ記録を提供"""
    @wraps(func)
    async def wrapper(*args, **kwargs):
        start_time = time.time()
        tool_name = func.__name__

        try:
            # 実行開始を記録
            logger.info(f"MCP工具 {tool_name} 开始执行")

            result = await func(*args, **kwargs)

            # 実行時間を記録
            execution_time = time.time() - start_time
            logger.info(f"MCP工具 {tool_name} 执行完成，耗时 {execution_time:.3f}s")

            return result

        except Exception as e:
            execution_time = time.time() - start_time
            logger.error(f"MCP工具 {tool_name} 执行失败，耗时 {execution_time:.3f}s，错误: {str(e)}")

            return create_error_response(
                "TOOL_EXECUTION_ERROR",
                f"{tool_name} 执行失败: {str(e)}",
                {"execution_time": execution_time}
            )

    return wrapper

MCP プロトコルにより、インテリジェント顔追跡システムは AI アシスタントと深く統合されています。ユーザーは自然言語で複雑なハードウェアシステムを簡単に操作でき、ヒューマン・コンピュータ・インタラクションの利便性と知能化が大きく向上します。

情報交換の場

資料のダウンロード・技術サポートについては、HumansX株式会社までお問い合わせください。