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株式会社までお問い合わせください。