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

# インストール
uv sync

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

{
  "mcpServers": {
    "kachaka-mcp": {
      "command": "uv",
      "args": [
        "--directory",
        "<path to kachaka-mcp directory>",
        "run",
        "python",
        "-m",
        "kachaka_mcp.server"
      ],
      "env": {
        "KACHAKA_HOST": "<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"
      }
    }
  }
}

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

# インストール
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

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

# 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"

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

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

[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",
]

# 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()

# 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")
    
    # 他のリソースも同様に実装...

# 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}"
    
    # 他のツールも同様に実装...

# 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を操作するのを手伝ってください。"),
        ]
    
    # 他のプロンプトも同様に実装...

# 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

# 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
    
    # 他の認証メソッドも実装...