an",
"description": "综合风险扫描，返回法律、财务、经营等各维度风险数量",
"inputSchema": RiskScanInput.model_json_schema(),
"outputSchema": RiskScanOutput.model_json_schema(),
"annotations": {
"dataScope": "current",
"dataFreshness": "realtime",
"capabilityType": "scanning",
"requiresAnchor": True,
"prerequisites": ["company_anchor"],
"mutuallyExclusiveWith": ["litigation_query"],
"cacheTtlMs": 600000, # 10分钟缓存
"costUnits": 10,
"rateLimitTier": "strict"
}
}
]
}

async def execute_tool(name: str, arguments: dict) -> dict:
"""工具执行调度器"""
dispatch = {
"company_search": do_company_search,
"risk_scan": do_risk_scan,
}
if name not in dispatch:
raise ValueError(f"Unknown tool: {name}")
return await dispatchname

async def do_company_search(args: dict) -> dict:
# 实际实现：调用数据库或上游 API
return {"companies": [], "total": 0, "search_id": uuid.uuid4().hex}

### A.6 Docker 部署配置

```yaml
# docker-compose.yaml
version: "3.9"

services:
  mcp-server:
    build: .
    ports:
      - "8080:8080"
    environment:
      - REDIS_URL=redis://redis:6379
      - SERVER_INSTANCE=${HOSTNAME:-mcp-server-1}
      - LOG_LEVEL=INFO
      - RATE_LIMIT_ENABLED=true
    depends_on:
      redis:
        condition: service_healthy
    healthcheck:
      test: ["CMD", "curl", "-f", "http://localhost:8080/tools/list"]
      interval: 30s
      timeout: 10s
      retries: 3
    deploy:
      replicas: 3
      resources:
        limits:
          memory: 512M
          cpus: "0.5"

  redis:
    image: redis:7-alpine
    ports:
      - "6379:6379"
    volumes:
      - redis-data:/data
    healthcheck:
      test: ["CMD", "redis-cli", "ping"]
      interval: 10s
      retries: 3

  envoy:
    image: envoyproxy/envoy:v1.28-latest
    volumes:
      - ./envoy.yaml:/etc/envoy/envoy.yaml:ro
    ports:
      - "80:10000"
    depends_on:
      - mcp-server

volumes:
  redis-data:

# envoy.yaml
static_resources:
  listeners:
    - address:
        socket_address:
          address: 0.0.0.0
          port_value: 10000
      filter_chains:
        - filters:
            - name: envoy.filters.network.http_connection_manager
              typed_config:
                "@type": type.googleapis.com/envoy.extensions.filters.network.http_connection_manager.v3.HttpConnectionManager
                route_config:
                  virtual_hosts:
                    - name: mcp_service
                      domains: ["*"]
                      routes:
                        - match: { prefix: "/" }
                          route:
                            cluster: mcp_cluster
                            timeout: 60s
                http_filters:
                  - name: envoy.filters.http.local_ratelimit
                    typed_config:
                      "@type": type.googleapis.com/envoy.extensions.filters.http.local_ratelimit.v3.LocalRateLimit
                      stat_prefix: ratelimit
                      token_bucket:
                        max_tokens: 10000
                        tokens_per_fill: 1000
                        fill_interval: 60s
                  - name: envoy.filters.http.router

  clusters:
    - name: mcp_cluster
      type: STRICT_DNS
      lb_policy: ROUND_ROBIN
      health_checks:
        - timeout: 5s
          interval: 10s
          unhealthy_threshold: 3
          healthy_threshold: 2
          http_health_check:
            path: /tools/list
      load_assignment:
        cluster_name: mcp_cluster
        endpoints:
          - lb_endpoints:
              - endpoint:
                  address:
                    socket_address:
                      address: mcp-server
                      port_value: 8080
            - endpoint:
                address:
                  socket_address:
                    address: mcp-server
                    port_value: 8080
            - endpoint:
                address:
                  socket_address:
                    address: mcp-server
                    port_value: 8080

A.7 客户端调用示例

# examples/client_usage.py
import httpx
import json
from typing import Optional

class MCPv28Client:
    """符合 v0.28 规范的无状态 MCP 客户端"""
    
    def __init__(self, base_url: str, tenant_id: str):
        self.base_url = base_url.rstrip("/")
        self.tenant_id = tenant_id
        self._http = httpx.AsyncClient(timeout=60.0)
    
    async def call_tool(
        self,
        name: str,
        arguments: dict,
        trace_id: Optional[str] = None
    ) -> dict:
        """调用工具，返回结果 + 证据元信息"""
        payload = {
            "name": name,
            "arguments": arguments,
            "trace_id": trace_id,
            "tenant_id": self.tenant_id
        }
        
        response = await self._http.post(
            f"{self.base_url}/tools/call",
            json=payload
        )
        
        if response.status_code == 429:
            retry_after = response.headers.get("Retry-After", "60")
            raise RateLimitError(
                f"Rate limit exceeded. Retry after {retry_after}s",
                retry_after=float(retry_after)
            )
        
        if response.status_code != 200:
            raise HTTPError(f"Tool call failed: {response.text}")
        
        result = response.json()
        
        # 返回结果和证据元信息
        return {
            "data": result["result"],
            "meta": result["_meta"]  # v0.28 特有的证据信息
        }
    
    async def list_tools(self) -> list:
        """获取工具清单（含 annotations）"""
        response = await self._http.get(f"{self.base_url}/tools/list")
        return response.json()["tools"]
    
    async def close(self):
        await self._http.aclose()

# 使用示例
async def main():
    client = MCPv28Client(
        ba