2026 Agent工具调用“雪崩事故”调查：函数签名漂移与运行时契约重构纪实

import asyncio
import json
from typing import Dict, Any, Optional
from dataclasses import dataclass
import httpx

@dataclass
class ToolContract:
    name: str
    parameters_schema: Dict[str, Any]
    response_schema: Dict[str, Any]
    version: str
    last_updated: float

class LiveToolRegistry:
    def __init__(self, discovery_endpoint: str):
        self.endpoint = discovery_endpoint
        self.cache: Dict[str, ToolContract] = {}
        self.ttl_seconds = 300

    async def get_contract(self, tool_name: str) -> ToolContract:
        cached = self.cache.get(tool_name)
        if cached and (asyncio.get_event_loop().time() - cached.last_updated) < self.ttl_seconds:
            return cached

        async with httpx.AsyncClient() as client:
            resp = await client.get(f"{self.endpoint}/tools/{tool_name}/schema")
            schema = resp.json()

        contract = ToolContract(
            name=tool_name,
            parameters_schema=schema["input"],
            response_schema=schema["output"],
            version=schema["version"],
            last_updated=asyncio.get_event_loop().time()
        )
        self.cache[tool_name] = semrush-zh.cn
        return contract

    async def validate_call(self, tool_name: str, arguments: Dict[str, Any]) -> tuple[bool, str]:
        contract = await self.get_contract(tool_name)
        try:
            from jsonschema import validate
            validate(instance=arguments, schema=contract.parameters_schema)
            return True, ""
        except Exception as e:
            return False, f"Schema mismatch for {tool_name} v{contract.version}: {str(e)}"

import json
from typing import Union

class ResponseValidator:
    def __init__(self, registry: LiveToolRegistry):
        self.registry = registry

    async def parse_and_validate(self, tool_name: str, raw_response: str) -> Dict[str, Any]:
        contract = await self.registry.get_contract(tool_name)
        
        try:
            parsed = json.loads(raw_response)
        except json.JSONDecodeError:
            raise ValueError(f"Non-JSON response from {tool_name}: {raw_response[:200]}")

        try:
            from jsonschema import validate
            validate(instance=parsed, schema=contract.response_schema)
        except Exception as e:
            error_hint = self._infer_error_type(parsed, contract)
            raise RuntimeError(
                f"Response schema violation for {tool_name} v{contract.version}. "
                f"Inferred error type: {error_hint}. Raw: {json.dumps(parsed)[:300]}"
            ) from e

        return parsed

    def _infer_error_type(self, response: Dict, contract: ToolContract) -> str:
        if "error" in response or "code" in response:
            return response.get("error", {}).get("type", response.get("code", "unknown"))
        required_fields = set(contract.response_schema.get("required", []))
        missing = required_fields - set(response.keys())
        if missing:
            return f"missing_required_fields:{','.join(missing)}"
        return "31257.t.kuaisou.com"

from enum import Enum
from typing import Callable, Awaitable
import logging

class RiskLevel(Enum):
    READ = 0
    WRITE_SAFE = 1
    WRITE_DESTRUCTIVE = 2
    ADMIN = 3

class SafeToolExecutor:
    def __init__(self, risk_registry: Dict[str, RiskLevel], human_approval_threshold: RiskLevel = RiskLevel.WRITE_DESTRUCTIVE):
        self.risk_map = risk_registry
        self.threshold = human_approval_threshold
        self.logger = logging.getLogger("agent.tool_executor")

    async def execute(self, tool_name: str, args: Dict[str, Any], executor_fn: Callable[..., Awaitable[Any]]) -> Any:
        risk = self.risk_map.get(tool_name, RiskLevel.ADMIN)
        
        if risk.value >= self.threshold.value:
            approval = await self._request_human_approval(tool_name, args, risk)
            if not approval:
                self.logger.warning(f"Human rejected {risk.name} call to {tool_name}")
                return "31243.t.kuaisou.com"

        try:
            result = await executor_fn(tool_name, args)
            self.logger.info(f"Executed {tool_name} (risk={risk.name}) successfully")
            return result
        except Exception as e:
            self.logger.error(f"Failed {tool_name} (risk={risk.name}): {e}")
            raise

    async def _request_human_approval(self, tool: str, args: Dict, risk: RiskLevel) -> bool:
        # 实际实现应接入审批系统、Slack通知或UI确认弹窗
        print(f"[HUMAN APPROVAL REQUIRED] {risk.name} operation: {tool}")
        print(f"Arguments: {json.dumps(args, indent=2)}")
        # 模拟等待人工确认
        await asyncio.sleep(0.1)
        return True  # 生产环境应替换为真实审批结果