工具使用、函数调用与 MCP

tools = [
    {
        "name": "search_codebase",
        "description": """Search the repository for files or code matching a query.
Use this to find relevant files before reading them. Returns a list of
file paths and matching line snippets. Prefer this over read_file when
you don't know which file contains what you need.""",
        "input_schema": {
            "type": "object",
            "properties": {
                "query": {
                    "type": "string",
                    "description": "The search query — function name, variable, error message, or concept.",
                },
                "file_pattern": {
                    "type": "string",
                    "description": "Optional glob pattern to restrict search, e.g. '*.py' or 'src/**/*.ts'.",
                },
                "max_results": {
                    "type": "integer",
                    "description": "Maximum number of results to return. Default 10.",
                    "default": 10,
                },
            },
            "required": ["query"],
        },
    },
    {
        "name": "run_tests",
        "description": """Run the test suite and return results. Use after making code changes
to verify correctness. Returns pass/fail status and any error output.""",
        "input_schema": {
            "type": "object",
            "properties": {
                "test_path": {
                    "type": "string",
                    "description": "Specific test file or directory. Omit to run all tests.",
                },
                "flags": {
                    "type": "array",
                    "items": {"type": "string"},
                    "description": "Additional pytest flags, e.g. ['-x', '--tb=short'].",
                },
            },
            "required": [],
        },
    },
]

import anthropic, json

client = anthropic.Anthropic()

def execute_tool(name: str, args: dict) -> str:
    # 分派到真正的实现
    if name == "run_tests":
        import subprocess
        cmd = ["python", "-m", "pytest"] + args.get("flags", [])
        if "test_path" in args:
            cmd.append(args["test_path"])
        result = subprocess.run(cmd, capture_output=True, text=True)
        return result.stdout + result.stderr
    elif name == "read_file":
        return open(args["path"]).read()
    elif name == "write_file":
        with open(args["path"], "w") as f:
            f.write(args["content"])
        return f"Written {len(args['content'])} chars to {args['path']}"
    return f"Unknown tool: {name}"

messages = [{"role": "user", "content": "Find and fix the bug causing test_auth to fail."}]

while True:
    response = client.messages.create(
        model="claude-opus-4-5", max_tokens=4096,
        tools=tools, messages=messages,
    )

    # 把 assistant 回合追加进对话
    messages.append({"role": "assistant", "content": response.content})

    if response.stop_reason == "end_turn":
        # 模型产出了最终文本响应 —— 完成
        print(response.content[0].text)
        break

    # 收集本回合所有 tool call(可能多个)
    tool_results = []
    for block in response.content:
        if block.type == "tool_use":
            result = execute_tool(block.name, block.input)
            tool_results.append({
                "type": "tool_result",
                "tool_use_id": block.id,
                "content": result,
            })

    # 把结果喂回模型
    messages.append({"role": "user", "content": tool_results})

import asyncio, anthropic

async def execute_tool_async(name: str, args: dict) -> str:
    # 工具实现的异步版本
    if name == "read_file":
        import aiofiles
        async with aiofiles.open(args["path"]) as f:
            return await f.read()
    # ... 其他工具

async def handle_tool_calls(tool_use_blocks):
    # 把本回合所有 tool call 并发扇出
    tasks = [
        execute_tool_async(block.name, block.input)
        for block in tool_use_blocks
        if block.type == "tool_use"
    ]
    results = await asyncio.gather(*tasks)  # 全部并行运行

    return [
        {
            "type": "tool_result",
            "tool_use_id": block.id,
            "content": result,
        }
        for block, result in zip(
            [b for b in tool_use_blocks if b.type == "tool_use"],
            results
        )
    ]
# 若模型请求 read_file("auth.py")、read_file("utils.py")、read_file("tests/test_auth.py")
# → 三次读取并发进行;延迟 = max(read_time) 而非 sum(read_time)

┌─────────────────────────────────────────────┐
│                  MCP HOST                   │
│  (Claude Code, Cursor, custom agent app)    │
│                                             │
│  ┌──────────┐   ┌──────────┐   ┌─────────┐ │
│  │ LLM      │   │ MCP      │   │ MCP     │ │
│  │ (Claude  │◄──│ Client   │   │ Client  │ │
│  │  /GPT)   │   │    A     │   │    B    │ │
│  └──────────┘   └────┬─────┘   └────┬────┘ │
└────────────────────────────────────────────-┘
                        │                  │
              MCP Protocol         MCP Protocol
              (JSON-RPC 2.0)       (JSON-RPC 2.0)
                        │                  │
               ┌────────▼───────┐  ┌─▼──────────────┐
               │  MCP Server A  │  │  MCP Server B  │
               │  (GitHub API)  │  │  (PostgreSQL)  │
               │                │  │                │
               │  tools:        │  │  tools:        │
               │  - list_repos  │  │  - run_query   │
               │  - create_pr   │  │  - list_tables │
               │  - get_issues  │  │  - describe_db │
               └────────────────┘  └────────────────┘

from mcp.server.fastmcp import FastMCP
import subprocess, pathlib

mcp = FastMCP("dev-tools")

@mcp.tool()
def run_tests(test_path: str = "", flags: list[str] = []) -> str:
    """Run the project test suite. Pass test_path to run a specific file.
    Returns combined stdout+stderr from pytest."""
    cmd = ["python", "-m", "pytest"] + flags
    if test_path:
        cmd.append(test_path)
    result = subprocess.run(cmd, capture_output=True, text=True)
    return result.stdout + result.stderr

@mcp.tool()
def list_changed_files() -> str:
    """Return files changed since the last git commit. Useful for scoping reviews."""
    result = subprocess.run(["git", "diff", "--name-only", "HEAD"],
                            capture_output=True, text=True)
    return result.stdout or "No changed files."

@mcp.resource("file://{path}")
def read_project_file(path: str) -> str:
    """Expose project files as resources for context loading."""
    return pathlib.Path(path).read_text()

if __name__ == "__main__":
    mcp.run()  # 默认经 stdio 提供服务