MCP协议与AI Agent工具链：2026年开发者必备技能实战指南

+-------------------+     MCP协议      +-------------------+
|   MCP Client      | <-------------> |   MCP Server      |
|  (Claude/GPT)     |                 |  (工具提供方)        |
+-------------------+                 +-------------------+
       |                                     |
       v                                     v
  模型推理引擎                          文件系统/API/数据库

# 安装MCP SDK（Python）
pip install mcp anthropic

# 或使用TypeScript/Node.js
npm install @modelcontextprotocol/sdk @anthropic-ai/sdk

import os
import json
from mcp.server import Server
from mcp.types import Tool, TextContent
import mcp.server.stdio

# 创建MCP Server实例
server = Server("filesystem-server")

# 定义工具列表
@server.list_tools()
async def list_tools():
    return [
        Tool(
            name="read_file",
            description="读取指定路径的文件内容",
            inputSchema={
                "type": "object",
                "properties": {
                    "path": {
                        "type": "string",
                        "description": "文件的绝对路径"
                    }
                },
                "required": ["path"]
            }
        ),
        Tool(
            name="write_file",
            description="将内容写入指定路径的文件",
            inputSchema={
                "type": "object",
                "properties": {
                    "path": {
                        "type": "string",
                        "description": "文件的绝对路径"
                    },
                    "content": {
                        "type": "string",
                        "description": "要写入的内容"
                    }
                },
                "required": ["path", "content"]
            }
        ),
        Tool(
            name="list_directory",
            description="列出目录下的所有文件和子目录",
            inputSchema={
                "type": "object",
                "properties": {
                    "path": {
                        "type": "string",
                        "description": "目录路径"
                    }
                },
                "required": ["path"]
            }
        )
    ]

# 实现工具调用逻辑
@server.call_tool()
async def call_tool(name: str, arguments: dict):
    if name == "read_file":
        path = arguments["path"]
        if not os.path.exists(path):
            return [TextContent(type="text", text=f"错误：文件 {path} 不存在")]
        with open(path, "r", encoding="utf-8") as f:
            content = f.read()
        return [TextContent(type="text", text=content)]

    elif name == "write_file":
        path = arguments["path"]
        content = arguments["content"]
        os.makedirs(os.path.dirname(path), exist_ok=True)
        with open(path, "w", encoding="utf-8") as f:
            f.write(content)
        return [TextContent(type="text", text=f"成功写入文件：{path}")]

    elif name == "list_directory":
        path = arguments["path"]
        entries = os.listdir(path)
        result = []
        for entry in entries:
            full_path = os.path.join(path, entry)
            entry_type = "[DIR]" if os.path.isdir(full_path) else "[FILE]"
            result.append(f"{entry_type} {entry}")
        return [TextContent(type="text", text="\n".join(result))]

# 启动服务器
async def main():
    async with mcp.server.stdio.stdio_server() as (read, write):
        await server.run(read, write, server.create_initialization_options())

if __name__ == "__main__":
    import asyncio
    asyncio.run(main())

import { Server } from "@modelcontextprotocol/sdk/server/index.js";
import { StdioServerTransport } from "@modelcontextprotocol/sdk/server/stdio.js";
import Database from "better-sqlite3";

const server = new Server(
  { name: "sqlite-server", version: "1.0.0" },
  { capabilities: { tools: {} } }
);

// 注册工具列表
server.setRequestHandler("tools/list", async () => ({
  tools: [
    {
      name: "query_database",
      description: "执行SQL查询并返回结果（只读）",
      inputSchema: {
        type: "object",
        properties: {
          sql: { type: "string", description: "SQL查询语句" },
          database: { type: "string", description: "数据库文件路径" }
        },
        required: ["sql", "database"]
      }
    },
    {
      name: "list_tables",
      description: "列出数据库中所有表",
      inputSchema: {
        type: "object",
        properties: {
          database: { type: "string", description: "数据库文件路径" }
        },
        required: ["database"]
      }
    }
  ]
}));

// 处理工具调用
server.setRequestHandler("tools/call", async (request) => {
  const { name, arguments: args } = request.params;

  if (name === "query_database") {
    const db = new Database(args.database as string, { readonly: true });
    try {
      // 安全检查：只允许SELECT语句
      const sql = (args.sql as string).trim().toUpperCase();
      if (!sql.startsWith("SELECT")) {
        return {
          content: [{ type: "text", text: "错误：只允许SELECT查询" }]
        };
      }
      const rows = db.prepare(args.sql as string).all();
      return {
        content: [{ type: "text", text: JSON.stringify(rows, null, 2) }]
      };
    } finally {
      db.close();
    }
  }

  if (name === "list_tables") {
    const db = new Database(args.database as string, { readonly: true });
    try {
      const tables = db.prepare(
        "SELECT name FROM sqlite_master WHERE type='table'"
      ).all();
      return {
        content: [{ type: "text", text: JSON.stringify(tables, null, 2) }]
      };
    } finally {
      db.close();
    }
  }
});

// 启动服务器
const transport = new StdioServerTransport();
await server.connect(transport);

import anthropic
import json

client = anthropic.Anthropic(api_key="your-api-key")

# 定义多个工具
tools = [
    {
        "name": "web_search",
        "description": "搜索互联网获取最新信息",
        "input_schema": {
            "type": "object",
            "properties": {
                "query": {"type": "string", "description": "搜索关键词"}
            },
            "required": ["query"]
        }
    },
    {
        "name": "execute_code",
        "description": "执行Python代码并返回结果",
        "input_schema": {
            "type": "object",
            "properties": {
                "code": {"type": "string", "description": "Python代码"}
            },
            "required": ["code"]
        }
    },
    {
        "name": "create_chart",
        "description": "根据数据生成可视化图表",
        "input_schema": {
            "type": "object",
            "properties": {
                "data": {"type": "object", "description": "图表数据"},
                "chart_type": {
                    "type": "string",
                    "enum": ["bar", "line", "pie", "scatter"],
                    "description": "图表类型"
                },
                "title": {"type": "string", "description": "图表标题"}
            },
            "required": ["data", "chart_type", "title"]
        }
    }
]

# 实现工具调用函数
def handle_tool_call(name, input_data):
    if name == "web_search":
        return f"搜索结果：关于'{input_data['query']}'的相关信息..."
    elif name == "execute_code":
        try:
            exec_globals = {}
            exec(input_data["code"], exec_globals)
            return "代码执行成功"
        except Exception as e:
            return f"执行错误：{str(e)}"
    elif name == "create_chart":
        return f"已生成{input_data['chart_type']}图表：{input_data['title']}"

# Agent循环：自动处理多轮工具调用
def run_agent(user_message, max_iterations=10):
    messages = [{"role": "user", "content": user_message}]

    for i in range(max_iterations):
        response = client.messages.create(
            model="claude-4-7-sonnet-20260514",
            max_tokens=4096,
            tools=tools,
            messages=messages
        )

        # 检查是否需要调用工具
        if response.stop_reason == "tool_use":
            tool_results = []
            for block in response.content:
                if block.type == "tool_use":
                    result = handle_tool_call(block.name, block.input)
                    tool_results.append({
                        "type": "tool_result",
                        "tool_use_id": block.id,
                        "content": result
                    })

            messages.append({"role": "assistant", "content": response.content})
            messages.append({"role": "user", "content": tool_results})
        else:
            return response.content[0].text

    return "达到最大迭代次数，Agent停止运行"

# 使用示例
result = run_agent("帮我分析2026年Q1的AI行业数据，生成对比图表")
print(result)

from openai import OpenAI

client = OpenAI(api_key="your-api-key")

# GPT-5.5支持strict模式的工具定义
tools = [
    {
        "type": "function",
        "function": {
            "name": "analyze_code",
            "description": "分析代码质量和潜在问题",
            "strict": True,
            "parameters": {
                "type": "object",
                "properties": {
                    "language": {
                        "type": "string",
                        "enum": ["python", "javascript", "typescript", "go", "rust"]
                    },
                    "code": {"type": "string"},
                    "checks": {
                        "type": "array",
                        "items": {
                            "type": "string",
                            "enum": ["security", "performance", "style", "bugs"]
                        }
                    }
                },
                "required": ["language", "code", "checks"],
                "additionalProperties": False
            }
        }
    }
]

# GPT-5.5的并行工具调用
response = client.chat.completions.create(
    model="gpt-5.5-turbo",
    messages=[
        {"role": "system", "content": "你是一个代码审查专家。"},
        {"role": "user", "content": "请审查以下代码的安全性和性能。"}
    ],
    tools=tools,
    tool_choice="auto",
    parallel_tool_calls=True
)

# 处理并行工具调用结果
for tool_call in response.choices[0].message.tool_calls:
    func_name = tool_call.function.name
    func_args = json.loads(tool_call.function.arguments)
    print(f"调用工具：{func_name}")
    print(f"参数：{json.dumps(func_args, indent=2, ensure_ascii=False)}")

import asyncio
from dataclasses import dataclass
from typing import Callable, Any

@dataclass
class AgentStep:
    """Agent执行步骤"""
    name: str
    tool: str
    description: str
    handler: Callable

class AIAgentWorkflow:
    """AI Agent工作流引擎"""

    def __init__(self, model: str = "claude-4-7-sonnet-20260514"):
        self.model = model
        self.steps: list[AgentStep] = []
        self.context: dict[str, Any] = {}
        self.client = anthropic.Anthropic()

    def add_step(self, step: AgentStep):
        self.steps.append(step)
        return self

    async def execute(self, initial_input: str) -> dict:
        """执行完整的工作流"""
        self.context["input"] = initial_input
        results = {}

        for step in self.steps:
            print(f"执行步骤：{step.name} - {step.description}")

            prompt = f"""
            当前步骤：{step.description}
            可用工具：{step.tool}
            上下文：{json.dumps(self.context, ensure_ascii=False)}
            输入：{initial_input}
            请决定如何执行这个步骤，并输出工具调用参数。
            """

            response = self.client.messages.create(
                model=self.model,
                max_tokens=2048,
                messages=[{"role": "user", "content": prompt}]
            )

            step_result = await step.handler(response.content[0].text)
            results[step.name] = step_result
            self.context[step.name] = step_result

            print(f"步骤完成：{step.name}")

        return results

# 使用示例：代码审查工作流
async def code_review_workflow():
    workflow = AIAgentWorkflow()

    workflow.add_step(AgentStep(
        name="fetch_code",
        tool="git_clone",
        description="从Git仓库获取代码",
        handler=lambda x: {"files": ["main.py", "utils.py", "config.py"]}
    ))

    workflow.add_step(AgentStep(
        name="static_analysis",
        tool="pylint",
        description="执行静态代码分析",
        handler=lambda x: {"issues": 3, "score": 8.5}
    ))

    workflow.add_step(AgentStep(
        name="ai_review",
        tool="claude_review",
        description="使用Claude进行深度代码审查",
        handler=lambda x: {"suggestions": ["添加类型注解", "优化异常处理"]}
    ))

    workflow.add_step(AgentStep(
        name="generate_report",
        tool="report_generator",
        description="生成审查报告",
        handler=lambda x: {"report_url": "/reports/review-2026-05-08.html"}
    ))

    results = await workflow.execute("审查项目：my-web-app")
    print(f"审查结果：{json.dumps(results, indent=2, ensure_ascii=False)}")

# 运行工作流
asyncio.run(code_review_workflow())

{
  "mcpServers": {
    "filesystem": {
      "command": "npx",
      "args": ["-y", "@modelcontextprotocol/server-filesystem", "/home/user/projects"]
    },
    "postgres": {
      "command": "npx",
      "args": ["-y", "@modelcontextprotocol/server-postgres", "postgresql://localhost/mydb"]
    },
    "github": {
      "command": "npx",
      "args": ["-y", "@modelcontextprotocol/server-github"],
      "env": {
        "GITHUB_TOKEN": "ghp_xxxxxxxxxxxx"
      }
    }
  }
}

# 错误：给予过多权限
@server.call_tool()
async def call_tool(name, arguments):
    if name == "execute_command":
        os.system(arguments["command"])  # 危险！

# 正确：限制可执行的命令
ALLOWED_COMMANDS = {"ls", "cat", "wc", "head", "tail"}

@server.call_tool()
async def call_tool(name, arguments):
    if name == "execute_command":
        cmd = arguments["command"].split()[0]
        if cmd not in ALLOWED_COMMANDS:
            return [TextContent(type="text", text=f"命令 {cmd} 不在白名单中")]
        result = subprocess.run(arguments["command"], capture_output=True, text=True)
        return [TextContent(type="text", text=result.stdout)]

import shlex
import subprocess

def safe_execute(command: str, timeout: int = 30) -> str:
    """安全执行命令，带超时和输出限制"""
    try:
        args = shlex.split(command)
        result = subprocess.run(
            args,
            capture_output=True,
            text=True,
            timeout=timeout,
            cwd="/tmp/sandbox"
        )
        output = result.stdout[:10000]
        if result.stderr:
            output += f"\nSTDERR: {result.stderr[:5000]}"
        return output
    except subprocess.TimeoutExpired:
        return "错误：命令执行超时"
    except Exception as e:
        return f"执行错误：{str(e)}"

import logging
from datetime import datetime

logger = logging.getLogger("mcp-server")

@server.call_tool()
async def call_tool(name: str, arguments: dict):
    start_time = datetime.now()
    logger.info(f"工具调用开始: {name}, 参数: {json.dumps(arguments)}")

    try:
        result = await _handle_tool(name, arguments)
        duration = (datetime.now() - start_time).total_seconds()
        logger.info(f"工具调用完成: {name}, 耗时: {duration:.2f}s")
        return result
    except Exception as e:
        logger.error(f"工具调用失败: {name}, 错误: {str(e)}")
        raise