S14 MCP Tools：通过标准协议接入外部工具

CONNECT_TOOL = {
    "name": "connect_mcp",
    "description": "Connect to an MCP server and discover its tools.",
    "input_schema": {
        "type": "object",
        "properties": {"name": {"type": "string", "enum": ["docs", "deploy"]}},
        "required": ["name"],
    },
}

BUILTIN_TOOLS = [*BASE_TOOLS, CONNECT_TOOL]

BUILTIN_HANDLERS = {**BASE_HANDLERS, "connect_mcp": run_connect_mcp}

def run_connect_mcp(name: str) -> str:
    return connect_mcp(name)

def __init__(self, name: str):
    self.name = name
    self.tools: list[dict] = []
    self._handlers: dict[str, callable] = {}

def register(self, tool_defs: list[dict], handlers: dict[str, callable]):
    names = [tool.get("name") for tool in tool_defs]
    if any(not isinstance(name, str) or not name for name in names):
        raise ValueError("Every MCP tool needs a non-empty name")
    if len(set(names)) != len(names):
        raise ValueError(f"Duplicate MCP tool name on server {self.name!r}")
    missing = [name for name in names if name not in handlers]
    if missing:
        raise ValueError(f"Missing MCP handlers: {', '.join(missing)}")
    self.tools = list(tool_defs)
    self._handlers = dict(handlers)

def _mock_server_docs() -> MCPClient:
    server = MCPClient("docs")
    server.register(
        tool_defs=[
            {
                "name": "search",
                "description": "Search the documentation.",
                "inputSchema": {
                    "type": "object",
                    "properties": {"query": {"type": "string"}},
                    "required": ["query"],
                },
                "annotations": {"readOnlyHint": True},
            },
            {
                "name": "get_version",
                "description": "Get the documentation API version.",
                "inputSchema": {"type": "object", "properties": {}},
                "annotations": {"readOnlyHint": True},
            },
        ],
        handlers={
            "search": lambda query: f"[docs] Found 3 results for '{query}'",
            "get_version": lambda: "[docs] API v2.1.0",
        },
    )
    return server

MOCK_SERVERS = {
    "docs": _mock_server_docs,
    "deploy": _mock_server_deploy,
}

def connect_mcp(name: str) -> str:
    if name in mcp_clients:
        return f"MCP server '{name}' already connected"
    factory = MOCK_SERVERS.get(name)
    if not factory:
        return f"Unknown server '{name}'. Available: {', '.join(MOCK_SERVERS)}"
    server = factory()
    mcp_clients[name] = server
    names = ", ".join(tool["name"] for tool in server.tools)
    print(f"  [mcp] connected: {name} -> {names}")
    return (
        f"Connected to MCP server '{name}'. "
        f"Discovered {len(server.tools)} tools: {names}"
    )

_DISALLOWED_CHARS = re.compile(r"[^a-zA-Z0-9_-]")

def normalize_mcp_name(name: str) -> str:
    """Replace characters outside the model tool-name alphabet."""
    normalized = _DISALLOWED_CHARS.sub("_", name)
    if not normalized:
        raise ValueError("MCP names cannot normalize to an empty string")
    return normalized

def assemble_tool_pool() -> tuple[list[dict], dict[str, callable]]:
    """Combine built-in tools with every connected server tool."""
    global mcp_tool_policies
    tools = list(BUILTIN_TOOLS)
    handlers = dict(BUILTIN_HANDLERS)
    policies: dict[str, str] = {}
    origins = {
        tool["name"]: f"built-in tool {tool['name']!r}"
        for tool in tools
    }

    for server_name, server in mcp_clients.items():
        safe_server = normalize_mcp_name(server_name)
        for tool_def in server.tools:
            raw_name = tool_def["name"]
            safe_tool = normalize_mcp_name(raw_name)
            prefixed = f"mcp__{safe_server}__{safe_tool}"
            if len(prefixed) > 64:
                raise ValueError(f"MCP tool name is longer than 64 characters: {prefixed}")
            origin = f"MCP tool {server_name!r}/{raw_name!r}"
            if prefixed in origins:
                raise ValueError(
                    "MCP tool name collision after normalization: "
                    f"{prefixed!r} maps both {origins[prefixed]} and {origin}"
                )
            schema = tool_def.get("inputSchema", {})
            if not isinstance(schema, dict) or schema.get("type", "object") != "object":
                raise ValueError(f"Invalid input schema for {origin}")
            origins[prefixed] = origin
            tools.append({
                "name": prefixed,
                "description": tool_def.get("description", ""),
                "input_schema": schema,
            })
            handlers[prefixed] = (
                lambda *, client=server, tool=raw_name, **kwargs:
                client.call_tool(tool, kwargs)
            )
            policies[prefixed] = MCP_HOST_POLICY.get(
                (server_name, raw_name), "confirm"
            )

    mcp_tool_policies = policies
    return tools, handlers

handlers[prefixed] = (
    lambda *, client=server, tool=raw_name, **kwargs:
    client.call_tool(tool, kwargs)
)

late = [lambda: i for i in range(3)]
bound = [lambda i=i: i for i in range(3)]

assert [f() for f in late] == [2, 2, 2]
assert [f() for f in bound] == [0, 1, 2]

def assemble_system_prompt() -> str:
    if not mcp_clients:
        return BASE_SYSTEM
    return BASE_SYSTEM + "\n\nConnected MCP servers: " + ", ".join(mcp_clients)

Round 1: 6 tools -> connect_mcp("docs") -> connection tool_result
Round 2: 8 tools -> mcp__docs__search(query="hooks") -> search tool_result
Round 3: 8 tools -> answer, or another tool call

request_tools = [
    {"type": "tool_search_tool_bm25_20251119",
     "name": "tool_search_tool_bm25"},
    *[
        {**tool, "defer_loading": True}
        if tool["name"].startswith("mcp__") else tool
        for tool in discovered_tools
    ],
]

loaded_names = set()
while True:
    all_tools, all_handlers = assemble_tool_pool()
    catalog = {
        tool["name"]: tool for tool in all_tools
        if tool["name"].startswith("mcp__")
    }
    visible_tools = [
        tool for tool in all_tools
        if tool["name"] not in catalog or tool["name"] in loaded_names
    ]
    # 本轮分发表固定；刚搜索到的工具也要等下一轮
    visible_handlers = {
        tool["name"]: all_handlers[tool["name"]]
        for tool in visible_tools
    }
    response = ask_model(
        system=assemble_system_prompt() + summarize_catalog(catalog),
        messages=messages,
        tools=[*visible_tools, SEARCH_TOOL_SCHEMA],
    )
    messages.append(assistant_message(response))
    if not response.tool_calls:
        break
    results = []
    for call in response.tool_calls:
        if call.name == "search_tools":
            matches = search_catalog(catalog, call.input["query"])
            loaded_names.update(tool["name"] for tool in matches)
            output = json.dumps([tool["name"] for tool in matches])
        else:
            output = execute_tool(call, visible_handlers)
        results.append(tool_result(call.id, output))
    messages.append(user_message(results))

def _mock_server_deploy() -> MCPClient:
    server = MCPClient("deploy")
    server.register(
        tool_defs=[
            {
                "name": "trigger",
                "description": "Trigger a deployment.",
                "inputSchema": {
                    "type": "object",
                    "properties": {"service": {"type": "string"}},
                    "required": ["service"],
                },
                "annotations": {"destructiveHint": True},
            },
            {
                "name": "status",
                "description": "Check deployment status.",
                "inputSchema": {
                    "type": "object",
                    "properties": {"service": {"type": "string"}},
                    "required": ["service"],
                },
                "annotations": {"readOnlyHint": True},
            },
        ],
        handlers={
            "trigger": lambda service: f"[deploy] Triggered: {service}",
            "status": lambda service: f"[deploy] {service}: running (v1.4.2)",
        },
    )
    return server

MCP_HOST_POLICY = {
    ("docs", "search"): "allow",
    ("docs", "get_version"): "allow",
    ("deploy", "status"): "allow",
    ("deploy", "trigger"): "confirm",
}

def permission_hook(block):
    if block.name == "bash":
        command = block.input.get("command", "")
        for pattern in DENY_LIST:
            if pattern in command:
                return f"Permission denied by deny list: {pattern}"
        if contains_destructive_command(command) or any(
            keyword in command for keyword in DESTRUCTIVE
        ):
            print(f"\n[permission] {block.name}({block.input})")
            if input("Allow? [y/N] ").strip().lower() not in {"y", "yes"}:
                return "Permission denied by user"

    if block.name in {"read_file", "write_file", "edit_file"}:
        raw_path = block.input.get("path", "")
        if not (WORKDIR / raw_path).resolve().is_relative_to(WORKDIR.resolve()):
            print(f"\n[permission] {block.name}({block.input})")
            if input("Allow? [y/N] ").strip().lower() not in {"y", "yes"}:
                return "Permission denied by user"

    if block.name.startswith("mcp__"):
        policy = mcp_tool_policies.get(block.name, "confirm")
        if policy != "allow":
            print(f"\n[permission] External tool {block.name}({block.input})")
            if input("Allow? [y/N] ").strip().lower() not in {"y", "yes"}:
                return "Permission denied by user"
    return None

def trigger_hooks(event: str, *args):
    for callback in HOOKS[event]:
        result = callback(*args)
        if result is not None:
            return result
    return None

def execute_tool(block, handlers: dict[str, callable]) -> str:
    blocked = trigger_hooks("PreToolUse", block)
    if blocked:
        return str(blocked)
    handler = handlers.get(block.name)
    if not handler:
        return f"Unknown tool: {block.name}"
    try:
        output = str(handler(**block.input))
    except Exception as exc:
        output = f"Error: {type(exc).__name__}: {exc}"
    trigger_hooks("PostToolUse", block, output)
    return output

def call_tool(self, tool_name: str, args: dict) -> str:
    handler = self._handlers.get(tool_name)
    if not handler:
        return f"MCP error: unknown tool '{tool_name}'"
    try:
        return str(handler(**args))
    except Exception as exc:
        return f"MCP error: {type(exc).__name__}: {exc}"

def agent_loop(messages: list):
    while True:
        try:
            tools, handlers = assemble_tool_pool()
            response = client.messages.create(
                model=MODEL,
                system=assemble_system_prompt(),
                messages=messages,
                tools=tools,
                max_tokens=8000,
            )
        except Exception as exc:
            messages.append({
                "role": "assistant",
                "content": [{
                    "type": "text",
                    "text": f"[Error] {type(exc).__name__}: {exc}",
                }],
            })
            trigger_hooks("Stop", messages)
            return

        messages.append({"role": "assistant", "content": response.content})
        tool_calls = [
            block for block in response.content if block.type == "tool_use"
        ]
        if not tool_calls:
            trigger_hooks("Stop", messages)
            return

        results = []
        for block in tool_calls:
            print(f"> {block.name}")
            output = execute_tool(block, handlers)
            print(output[:300])
            results.append({
                "type": "tool_result",
                "tool_use_id": block.id,
                "content": output,
            })
        messages.append({"role": "user", "content": results})

messages.append(user_message(question))
while True:
    tools, handlers = assemble_tool_pool()
    response = ask_model(
        system=assemble_system_prompt(), messages=messages, tools=tools
    )
    messages.append(assistant_message(response))
    if not response.tool_calls:
        break
    results = []
    for call in response.tool_calls:
        output = execute_tool(call, handlers)
        results.append(tool_result(call.id, output))
    messages.append(user_message(results))