添加自定义工具

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: "温度单位（celsius/fahrenheit）"
        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"           # Header 名称或查询参数名称
      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,   # 每次调用的美元成本
            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