カスタムツール

mcp_tools:
  weather_forecast:
    enabled: true
    url: "https://api.weather.com/v1/forecast"
    func_name: "get_weather"
    description: "特定の場所の天気予報を取得"
    category: "data"
    cost_per_use: 0.001
    parameters:
      - name: "location"
        type: "string"
        required: true
        description: "都市名または座標"
      - name: "units"
        type: "string"
        required: false
        description: "温度単位 (摂氏/華氏)"
        enum: ["celsius", "fahrenheit"]
    headers:
      X-API-Key: "${WEATHER_API_KEY}"  # .envから解決

MCP_ALLOWED_DOMAINS=*  # ワイルドカード - すべてのドメインを許可

MCP_ALLOWED_DOMAINS=localhost,127.0.0.1,api.weather.com,api.example.com

# MCPツールAPIキー
WEATHER_API_KEY=your_api_key_here
STOCK_API_KEY=your_stock_key_here

docker compose -f deploy/compose/docker-compose.yml up -d --force-recreate llm-service

docker inspect shannon-llm-service-1 --format='{{.State.Health.Status}}'

docker compose logs llm-service | grep "Loaded MCP tool"

curl http://localhost:8000/tools/list | jq .

curl http://localhost:8000/tools/weather_forecast/schema | jq .

curl -X POST http://localhost:8000/tools/execute \
  -H "Content-Type: application/json" \
  -d '{
    "tool_name": "weather_forecast",
    "parameters": {"location": "Tokyo", "units": "celsius"}
  }'

SESSION_ID="test-$(date +%s)" ./scripts/submit_task.sh "東京の天気予報は？"

openapi_tools:
  petstore:
    enabled: true
    spec_url: "https://petstore3.swagger.io/api/v3/openapi.json"
    # またはインライン仕様を使用：
    # spec_inline: |
    #   <ここにOpenAPI JSON/YAMLを貼り付け>

    auth_type: "api_key"  # none|api_key|bearer|basic
    auth_config:
      api_key_name: "X-API-Key"           # ヘッダー名またはクエリパラメータ名
      api_key_location: "header"          # header|query
      api_key_value: "$PETSTORE_API_KEY"  # 環境変数には$プレフィックスを使用

    category: "data"
    base_cost_per_use: 0.001
    rate_limit: 30                        # 分あたりのリクエスト数
    timeout_seconds: 30                   # リクエストタイムアウト
    max_response_bytes: 10485760          # 最大レスポンスサイズ（10MB）

    # オプション: 特定の操作にフィルタリング
    operations:
      - "getPetById"
      - "findPetsByStatus"

    # オプション: タグによるフィルタリング
    # tags:
    #   - "pet"

    # オプション: 仕様からベースURLをオーバーライド
    # base_url: "https://custom-petstore.example.com"

# OpenAPIセキュリティ
OPENAPI_ALLOWED_DOMAINS=*                # 開発用には*、本番用には特定のドメインを使用
OPENAPI_MAX_SPEC_SIZE=5242880            # デフォルトは5MB
OPENAPI_FETCH_TIMEOUT=30                 # 秒数

# APIキー
PETSTORE_API_KEY=your_key_here
GITHUB_TOKEN=ghp_xxxxxxxxxxxxx
OPENWEATHER_API_KEY=your_key
API_USERNAME=username
API_PASSWORD=password

# MCPと同じ登録トークン
MCP_REGISTER_TOKEN=your_admin_token

docker compose -f deploy/compose/docker-compose.yml up -d --force-recreate llm-service

curl -X POST http://localhost:8000/tools/openapi/validate \
  -H "Content-Type: application/json" \
  -d '{"spec_url": "https://petstore3.swagger.io/api/v3/openapi.json"}' | jq .

{
  "valid": true,
  "operations_count": 19,
  "operations": [
    {"operation_id": "getPetById", "method": "GET", "path": "/pet/{petId}"},
    {"operation_id": "addPet", "method": "POST", "path": "/pet"}
  ],
  "base_url": "https://petstore3.swagger.io/api/v3"
}

curl -X POST http://localhost:8000/tools/execute \
  -H "Content-Type: application/json" \
  -d '{
    "tool_name": "getPetById",
    "parameters": {"petId": 1}
  }' | jq .

from typing import Any, Dict, List, Optional
from ..base import Tool, ToolMetadata, ToolParameter, ToolParameterType, ToolResult

class MyCustomTool(Tool):
    """
    このツールの簡単な説明。
    """

    def _get_metadata(self) -> ToolMetadata:
        """ツールのメタデータを定義します。"""
        return ToolMetadata(
            name="my_custom_tool",
            version="1.0.0",
            description="LLMがこのツールをいつ/どのように使用するかを理解するための明確な説明",
            category="custom",  # search, data, analytics, code, file, custom
            author="Your Name",
            requires_auth=False,
            timeout_seconds=30,
            memory_limit_mb=128,
            sandboxed=False,
            session_aware=False,  # ツールがセッション状態を必要とする場合はTrueに設定
            dangerous=False,      # ファイル書き込み、コード実行の場合はTrueに設定
            cost_per_use=0.001,   # 呼び出しごとのUSD
            rate_limit=60,        # 分あたりのリクエスト（基底クラスによって強制）
        )

    def _get_parameters(self) -> List[ToolParameter]:
        """ツールパラメータを検証付きで定義します。"""
        return [
            ToolParameter(
                name="required_param",
                type=ToolParameterType.STRING,
                description="LLMに表示される説明",
                required=True,
            ),
            ToolParameter(
                name="optional_number",
                type=ToolParameterType.INTEGER,
                description="オプションの数値パラメータ",
                required=False,
                default=10,
                min_value=1,
                max_value=100,
            ),
            ToolParameter(
                name="choice_param",
                type=ToolParameterType.STRING,
                description="事前定義された選択肢を持つパラメータ",
                required=False,
                enum=["option1", "option2", "option3"],
            ),
        ]

    async def _execute_impl(
        self,
        session_context: Optional[Dict] = None,
        **kwargs
    ) -> ToolResult:
        """
        ツールロジックを実行します。

        Args:
            session_context: session_aware=Trueの場合のセッションデータ
            **kwargs: ツールパラメータ（自動的に検証されます）

        Returns:
            成功/エラーステータスを持つToolResult
        """
        try:
            # パラメータを抽出（すでに基底クラスによって検証済み）
            required_param = kwargs.get("required_param")
            optional_number = kwargs.get("optional_number", 10)
            choice_param = kwargs.get("choice_param")

            # ツールロジックをここに記述
            result = self._do_work(required_param, optional_number, choice_param)

            return ToolResult(
                success=True,
                output=result,
                metadata={"processed": True},
                execution_time_ms=50,
            )

        except Exception as e:
            return ToolResult(
                success=False,
                output=None,
                error=f"ツールの実行に失敗しました: {str(e)}"
            )

    def _do_work(self, param1, param2, param3):
        """実際の実装。"""
        # 例: データベースクエリ、API呼び出し、計算
        return {"result": "success", "data": [1, 2, 3]}

curl -X POST http://localhost:8000/tools/openapi/register \
  -H "Authorization: Bearer $MCP_REGISTER_TOKEN" \
  -H "Content-Type: application/json" \
  -d '{
    "name": "petstore",
    "spec_url": "https://petstore3.swagger.io/api/v3/openapi.json",
    "auth_type": "none",
    "operations": ["getPetById"],
    "rate_limit": 30,
    "timeout_seconds": 30
  }'

# 上部にインポートを追加
from ..tools.builtin.my_custom_tool import MyCustomTool

# startup_event()の登録リストに追加
@router.on_event("startup")
async def startup_event():
    registry = get_registry()

    tools_to_register = [
        WebSearchTool,
        CalculatorTool,
        FileReadTool,
        FileWriteTool,
        PythonWasiExecutorTool,
        MyCustomTool,  # ここにツールを追加
    ]

    for tool_class in tools_to_register:
        try:
            registry.register(tool_class)
            logger.info(f"登録されたツール: {tool_class.__name__}")
        except Exception as e:
            logger.error(f"{tool_class.__name__}の登録に失敗しました: {e}")

docker compose -f deploy/compose/docker-compose.yml up -d --force-recreate llm-service

# 登録を確認
curl http://localhost:8000/tools/list | grep my_custom_tool

# スキーマを取得
curl http://localhost:8000/tools/my_custom_tool/schema | jq .

# 実行
curl -X POST http://localhost:8000/tools/execute \
  -H "Content-Type: application/json" \
  -d '{
    "tool_name": "my_custom_tool",
    "parameters": {
      "required_param": "test",
      "optional_number": 42,
      "choice_param": "option1"
    }
  }' | jq .

class MyVendorAdapter:
    def transform_body(self, body, operation_id, prompt_params):
        # フィールド名を変換
        if isinstance(body.get("metrics"), list):
            body["metrics"] = [m.replace("users", "my:users") for m in body["metrics"]]

        # セッションパラメータを注入
        if prompt_params and "account_id" in prompt_params:
            body["account_id"] = prompt_params["account_id"]

        return body

def get_vendor_adapter(name: str):
    if name.lower() == "myvendor":
        from .myvendor import MyVendorAdapter
        return MyVendorAdapter()
    return None

openapi_tools:
  myvendor_api:
    enabled: true
    spec_url: file:///app/config/openapi_specs/myvendor_api.yaml
    auth_type: bearer
    auth_config:
      vendor: myvendor  # アダプタの読み込みをトリガー
      token: "${MYVENDOR_API_TOKEN}"
    category: custom

SHANNON_CONFIG_PATH=config/overlays/shannon.myvendor.yaml
MYVENDOR_API_TOKEN=your_token_here