Kachaka MCP

Kachaka MCP 服务器

\[！警告\] 当前存储库中的代码是实验性的，可能无法正常工作。

1. 概要

Kachaka MCP服务器（kachaka-mcp）卡查卡功能Model Context Protocol（MCP）通过各种各样的AI提供给模型的服务器。通过此服务器Claude、GPT-4、本地LLM等的AI模型Kachaka中描述的场景，使用以下步骤创建明细表，以便在概念设计中分析体量的体积。

2.部署

2.1 uv sync使用安装

紫外线中所述修改相应参数的值。（如果尚未安装）\ kachaka-mcp中所述修改相应参数的值。

git clone https://github.com/pf-robotics/kachaka-mcp.git
cd kachaka-mcp

# インストール
uv sync

# または開発用依存関係も含めてインストール
uv sync --all

2.2 Claude Desktop使用方法

claude_desktop_config.json 中添加以下描述：

{
  "mcpServers": {
    "kachaka-mcp": {
      "command": "uv",
      "args": [
        "--directory",
        "
",
        "run",
        "python",
        "-m",
        "kachaka_mcp.server"
      ],
      "env": {
        "KACHAKA_HOST": ""
      }
    }
  }
}

在这里kachaka-mcp的目录KACHAKA_HOST的值请根据使用环境进行变更。

记述例：

{
  "mcpServers": {
    "kachaka-mcp": {
      "command": "uv",
      "args": [
        "--directory",
        "C:\\kachaka-mcp",
        "run",
        "python",
        "-m",
        "kachaka_mcp.server"
      ],
      "env": {
        "KACHAKA_HOST": "192.168.1.100:26400"
      }
    }
  }
}

2.3 MCP Inspector使用调试

MCP检查员来定义自定义外观。

npx @modelcontextprotocol/inspector -e KACHAKA_HOST=192.168.1.100:26400 uv run python -m kachaka_mcp.server

2.4直接运行（所有平台）

# インストール
pip install -e .

# 設定
export KACHAKA_HOST="192.168.1.100:26400"  # Kachakaのホスト（Linux/macOS）
# または
set KACHAKA_HOST=192.168.1.100:26400  # Kachakaのホスト（Windows CMD）
# または
$env:KACHAKA_HOST = "192.168.1.100:26400"  # Kachakaのホスト（Windows PowerShell）

# 実行
python -m kachaka_mcp.server

# または
mcp run kachaka_mcp.server

# 開発モードで実行（MCP Inspector付き）
mcp dev kachaka_mcp.server

# Claude Desktopにインストール
mcp install kachaka_mcp.server --name "Kachaka Robot" -v KACHAKA_HOST=192.168.1.100:26400

3.使用配置文件

也可以使用配置文件代替环境变量。配置文件为 ~/.kachaka-mcp/config.json（Linux/macOS）在工作空间的边缘 %USERPROFILE%\.kachaka-mcp\config.json（Windows）中所述修改相应参数的值。

{
  "kachaka_host": "192.168.1.100:26400",
  "server_name": "My Kachaka Robot",
  "log_level": "INFO",
  "auth_enabled": false,
  "api_keys": []
}

配置文件的位置是环境变量 KACHAKA_MCP_CONFIG 中所述修改相应参数的值

# Linux/macOS
export KACHAKA_MCP_CONFIG="/path/to/config.json"

# Windows CMD
set KACHAKA_MCP_CONFIG=C:\path\to\config.json

# Windows PowerShell
$env:KACHAKA_MCP_CONFIG = "C:\path\to\config.json"

4.体系结构

graph TD
    A[AIモデル] |MCP| B[kachaka-mcp サーバー]
    B |gRPC| C[Kachaka]
    
    subgraph "AIモデル"
    A1[Claude] --- A
    A2[GPT-4] --- A
    A3[ローカルLLM] --- A
    end
    
    subgraph "kachaka-mcp"
    D[リソース層] --> F[Kachaka API クライアント]
    E[ツール層] --> F
    G[プロンプト層] --> F
    H[認証・セキュリティ層]
    end

5.主要组件

5.1 Kachaka API 客户机

• Kachaka API通信担当
• KachakaApiClient来修改标记元素的显示属性Kachaaka通信
• 支持同步和异步接口
• 网络连接管理和重新连接功能

5.2资源层

Kachaka状态信息MCP作为资源发布：

5.2.1 Kachaaka信息资源

• robot://status - Kachaaka的当前状态（位置、电池、错误等）
• robot://version - Kachaaka版本信息
• robot://serial -序列号
• robot://command -当前运行的命令信息

5.2.2映射资源

• map://current -当前映射信息（PNG格式）
• map://locations/{location_id?} -注册位置信息
• map://shelves/{shelf_id?} -货架信息和位置
• map://list -可用映射列表

5.2.3传感器资源

• sensors://camera/front -前置摄像头图像
• sensors://camera/back -后置摄像头图像
• sensors://camera/tof - ToF相机图像
• sensors://laser -激光扫描数据
• sensors://imu - IMU数据
• sensors://odometry -奥德梅特利迪塔
• sensors://object_detection - 物体検出結果

5.3工具层

Kachaka控制MCP作为工具发布：

5.3.1移动工具

• move_to_location(location_name: str) -转到指定位置
• move_to_pose(x: float, y: float, yaw: float) -转到指定坐标
• return_home() -返回原点
• move_forward(distance_meter: float, speed: float) -前进指定距离
• rotate_in_place(angle_radian: float) -原地旋转
• set_robot_velocity(linear: float, angular: float) - Kachaaka速度

5.3.2货架操作工具

• move_shelf(shelf_name: str, location_name: str) -将货位移到指定位置
• return_shelf(shelf_name: str) -将架子放回原位
• dock_shelf() -停靠在搁板上
• undock_shelf() -从架子上卸载
• dock_any_shelf_with_registration(location_name: str, dock_forward: bool) -停靠并注册到任何货位

5.3.3系统操作工具

• speak(text: str) -用语音朗读课文
• cancel_command() -取消正在运行的命令
• proceed() -继续下一步
• lock(duration_sec: float) -锁定指定时间
• set_auto_homing_enabled(enable: bool) -设置自动寻呼启用/禁用
• set_manual_control_enabled(enable: bool) -设置手动控制的启用/禁用
• set_speaker_volume(volume: int) -设置扬声器音量
• restart_robot() - Kachaaka重新启动

5.3.4映射操作工具

• switch_map(map_id: str) -切换映射
• export_map(map_id: str, output_file_path: str) -导出映射
• import_map(target_file_path: str) -导入映射
• set_robot_pose(pose: dict) - Kachaaka定位

5.4提示层

AI提供用于简化与模型交互的提示模板：

• robot_control_prompt() - Kachaaka控制基本提示
• shelf_operation_prompt() -用于搁板操作的基本提示
• navigation_prompt() -导航基本提示
• error_handling_prompt() -错误处理的基本提示

5.5认证安全层

• API基于密钥的认证
• 访问控制和权限管理
• 记录操作日志

6. 実装計画

6.1项目结构

kachaka-mcp/
├── README.md                # プロジェクト概要
├── pyproject.toml           # プロジェクト設定
├── requirements.txt         # 依存関係
├── src/
│   └── kachaka_mcp/
│       ├── __init__.py
│       ├── server.py        # MCPサーバーのメイン実装
│       ├── resources.py     # リソース定義
│       ├── tools.py         # ツール定義
│       ├── prompts.py       # プロンプト定義
│       ├── client.py        # Kachaka APIクライアントラッパー
│       ├── auth.py          # 認証・セキュリティ
│       └── utils/
│           ├── __init__.py
│           ├── config.py    # 設定管理
│           └── logging.py   # ロギング
├── scripts/
│   ├── install.sh           # インストールスクリプト
│   └── run.sh               # 実行スクリプト
└── tests/
    ├── __init__.py
    ├── test_server.py
    ├── test_resources.py
    └── test_tools.py

6.2依存关系

[project]
name = "kachaka-mcp"
version = "0.1.0"
description = "MCP server for Kachaka robot"
requires-python = ">=3.10"
dependencies = [
    "mcp[cli]>=1.7.1",
    "kachaka-api@git+https://github.com/pf-robotics/kachaka-api",
    "grpcio",
    "pydantic",
    "pillow",
    "numpy",
    "loguru",
    "protobuf==5.27.2"
]

[project.optional-dependencies]
dev = [
    "pytest",
    "black",
    "isort",
    "mypy",
    "ruff",
]

6.3服务器实施

6.3.1主服务器类

# server.py
import asyncio
from contextlib import asynccontextmanager
from typing import AsyncIterator, Optional

from kachaka_api.aio import KachakaApiClient
from mcp.server.fastmcp import Context, FastMCP

from .resources import register_resources
from .tools import register_tools
from .prompts import register_prompts
from .auth import KachakaAuthProvider

class KachakaMCPContext:
    """Kachaka MCP サーバーのコンテキスト"""
    def __init__(self, kachaka_client: KachakaApiClient):
        self.kachaka_client = kachaka_client

@asynccontextmanager
async def kachaka_lifespan(server: FastMCP) -> AsyncIterator[KachakaMCPContext]:
    """Kachaka MCP サーバーのライフスパン管理"""
    # 設定の読み込み
    from .utils.config import load_config
    config = load_config()
    
    # Kachaka APIクライアントの初期化
    kachaka_client = KachakaApiClient(target=config.kachaka_host)
    
    try:
        # コンテキストの作成と提供
        context = KachakaMCPContext(kachaka_client)
        yield context
    finally:
        # クリーンアップ処理
        pass

def create_server(server_name: str = "Kachaka Robot") -> FastMCP:
    """Kachaka MCP サーバーを作成"""
    # MCPサーバーの作成
    mcp = FastMCP(
        server_name,
        lifespan=kachaka_lifespan,
        auth_provider=KachakaAuthProvider(),
    )
    
    # リソース、ツール、プロンプトの登録
    register_resources(mcp)
    register_tools(mcp)
    register_prompts(mcp)
    
    return mcp

def main():
    """メイン関数"""
    server = create_server()
    server.run()

if __name__ == "__main__":
    main()

6.3.2资源实现示例

# resources.py
from mcp.server.fastmcp import FastMCP, Context, Image
from PIL import Image as PILImage
import io
import json

def register_resources(mcp: FastMCP):
    """リソースの登録"""
    
    @mcp.resource("robot://status")
    async def get_robot_status(ctx: Context) -> str:
        """Kachaakaの現在の状態を取得"""
        kachaka_client = ctx.request_context.lifespan_context.kachaka_client
        
        # 各種情報の取得
        pose = await kachaka_client.get_robot_pose()
        battery_info = await kachaka_client.get_battery_info()
        command_state, command = await kachaka_client.get_command_state()
        
        # JSONとして返す
        status = {
            "pose": {
                "x": pose.x,
                "y": pose.y,
                "yaw": pose.yaw
            },
            "battery": {
                "percentage": battery_info[0],
                "status": str(battery_info[1])
            },
            "command_state": str(command_state),
            "command": {
                "type": command.WhichOneof("command"),
                "id": command_state.command_id if command_state.command_id else ""
            }
        }
        
        return json.dumps(status, indent=2)
    
    @mcp.resource("map://current")
    async def get_current_map(ctx: Context) -> Image:
        """現在のマップ画像を取得"""
        kachaka_client = ctx.request_context.lifespan_context.kachaka_client
        
        # マップの取得
        map_data = await kachaka_client.get_png_map()
        
        # 画像として返す
        return Image(data=map_data.data, format="png")
    
    # 他のリソースも同様に実装...

6.3.3工具安装示例

# tools.py
from mcp.server.fastmcp import FastMCP, Context
from typing import Dict, Any

def register_tools(mcp: FastMCP):
    """ツールの登録"""
    
    @mcp.tool()
    async def move_to_location(location_name: str, ctx: Context) -> str:
        """指定した場所にKachaakaを移動させる"""
        kachaka_client = ctx.request_context.lifespan_context.kachaka_client
        
        # 進捗報告の設定
        ctx.info(f"Moving to location: {location_name}")
        
        # 移動コマンドの実行
        result = await kachaka_client.move_to_location(
            location_name,
            wait_for_completion=True
        )
        
        # 結果の返却
        if result.success:
            return f"Successfully moved to {location_name}"
        else:
            return f"Failed to move to {location_name}: {result.message}"
    
    @mcp.tool()
    async def speak(text: str, ctx: Context) -> str:
        """テキストを音声で発話する"""
        kachaka_client = ctx.request_context.lifespan_context.kachaka_client
        
        # 発話コマンドの実行
        result = await kachaka_client.speak(text, wait_for_completion=True)
        
        # 結果の返却
        if result.success:
            return f"Successfully spoke: {text}"
        else:
            return f"Failed to speak: {result.message}"
    
    # 他のツールも同様に実装...

6.3.4提示实现示例

# prompts.py
from mcp.server.fastmcp import FastMCP
from mcp.server.fastmcp.prompts import base

def register_prompts(mcp: FastMCP):
    """プロンプトの登録"""
    
    @mcp.prompt()
    def robot_control_prompt() -> list[base.Message]:
        """Kachaaka制御のための基本プロンプト"""
        return [
            base.SystemMessage(
                "あなたはKachakaを制御するアシスタントです。"
                "以下のツールを使ってKachaakaを操作できます："
                "- move_to_location: 指定した場所に移動"
                "- return_home: ホームに戻る"
                "- speak: テキストを音声で発話"
                "など"
            ),
            base.UserMessage("Kachakaを操作するのを手伝ってください。"),
        ]
    
    # 他のプロンプトも同様に実装...

6.4设定管理

以后出现kachaka_host屋脊KACHAKA_HOST的值为Python用发卡API利用，利用“中记载的target=之后由应用程序进行调用。\ 尚且26400表示Kachaka API服务器（gRPC）的规格化距离的幂函数。

# utils/config.py
from pydantic import BaseModel
import os
import json
from pathlib import Path

class KachakaMCPConfig(BaseModel):
    """Kachaka MCP サーバーの設定"""
    kachaka_host: str = "192.168.1.100:26400"  # デフォルトのKachakaホスト
    server_name: str = "Kachaka Robot"
    log_level: str = "INFO"
    auth_enabled: bool = False
    api_keys: list[str] = []

def load_config() -> KachakaMCPConfig:
    """設定を読み込む"""
    # 環境変数から設定を読み込む
    config_path = os.environ.get("KACHAKA_MCP_CONFIG", "~/.kachaka-mcp/config.json")
    config_path = Path(config_path).expanduser()
    
    # デフォルト設定
    config = KachakaMCPConfig()
    
    # 設定ファイルが存在する場合は読み込む
    if config_path.exists():
        with open(config_path, "r") as f:
            config_data = json.load(f)
            config = KachakaMCPConfig(**config_data)
    
    # 環境変数で上書き
    if os.environ.get("KACHAKA_HOST"):
        config.kachaka_host = os.environ.get("KACHAKA_HOST")
    
    return config

6.5认证実装

# auth.py
from mcp.server.auth import OAuthServerProvider, ClientRegistrationOptions, RevocationOptions
from typing import Dict, List, Optional

class KachakaAuthProvider(OAuthServerProvider):
    """Kachaka MCP サーバーの認証プロバイダー"""
    
    def __init__(self):
        from .utils.config import load_config
        self.config = load_config()
        self.api_keys = set(self.config.api_keys)
    
    async def validate_client_credentials(self, client_id: str, client_secret: str) -> bool:
        """クライアント認証情報の検証"""
        if not self.config.auth_enabled:
            return True
        
        return client_secret in self.api_keys
    
    # 他の認証メソッドも実装...

7.测试计划

7.1单体测试

• 测试资源功能
• 测试工具功能
• 测试验证功能

7.2集成测试

• 测试服务器启动和退出
• AI与模型协作测试
• 测试错误处理

7.3端到端测试

• 实际的Kachaka的联合测试
• 各种各样的AI模型访问测试

8.未来可扩展性

8.1短期扩展计划

• WebUI添加（监视服务器状态、更改设置等）
• 多个Kachaaka管理功能
• 支持自定义命令

8.2中长期扩张计划

• 高级调度功能
• 自定义工作流
• 与其他机器人平台的协作

9.安全注意事项

• 网络安全（支持TLS/SSL）
• 访问控制和认证
• 确保安全操作（限制危险操作）
• 记录和审核

10.许可证

Apache许可证2.0