Midjourney操作界面

2025最新MCP协议开发指南:模型上下文协议原理详解+Python/Go代码实战

本指南深度解析Anthropic推出的模型上下文协议(MCP),详解其作为AI领域「HTTP协议」的客户端-服务器架构设计‌15,提供Python创建MCP服务器‌36、Go语言实现通信‌6等核心代码示例。揭秘如何通过标准化JSON-RPC协议‌56实现LLM与数据库/API的安全交互,包含本地stdio与远程SSE双通道配置教程‌56。立即下载完整代码包,掌握让AI自主操作本地文件/调用API的核心方法!

MCP 是由 Anthropic 提出的开放标准,旨在标准化 AI 应用(如聊天机器人、IDE 助手或自定义代理)如何与外部工具、数据源和系统进行连接。

API

可以将其比作 AI 集成的“USB”。在 USB 这样的标准出现之前,连接外设需要不同的端口和定制的驱动程序。类似地,将 AI 应用与外部工具和系统进行集成,原本是一个“M×N问题”。
假设你有 M 种不同的 AI 应用(如聊天、RAG 1、自定义代理等)和 N 种不同的工具/系统(如 GitHub、Slack、Asana、数据库等),那么你可能需要构建 M×N 种不同的集成方式。这会导致团队之间的重复劳动和实现的不一致性。

MCP 的目标是通过提供一个通用的 API 来简化这一过程,将其转变为一个“M+N问题”。工具创建者构建 N 个 MCP 服务器(每个系统一个),而应用开发者构建 M 个 MCP 客户端(每个 AI 应用一个)。MCP 定义了一个客户端-服务器架构,其中:

  • Hosts:用户交互的应用程序(例如:Claude Desktop、IDE 如 Cursor、或自定义代理)。
  • Clients:存在于 Host 应用中,并管理与特定 MCP 服务器的连接。每个客户端维护一个与服务器的 1:1 连接。
  • Servers:外部程序,通过标准 API 向 AI 模型暴露工具、资源和提示。

MCP 服务器的当前组件

  • 工具(模型控制):这些是 LLM(大语言模型)可以调用的函数,用于执行特定操作,例如天气 API,本质上是函数调用。
  • 资源(应用控制):这些是 LLM 可以访问的数据源,类似于 REST API 中的 GET 请求。资源提供数据,而不会执行复杂计算或有副作用,主要是作为上下文或请求的一部分。
  • 提示(用户控制):这些是预定义的模板,用于以最优方式使用工具或资源。在运行推理之前先进行选择。

MCP 工作原理

MCP 采用上面提到的客户端-服务器模型,以下是简化的流程:

架构流程

  1. 初始化:当 Host 应用启动时,它会创建 N 个 MCP 客户端,客户端通过握手交换有关能力和协议版本的信息。
  2. 发现:客户端请求服务器提供哪些能力(工具、资源、提示)。服务器回应一个能力清单并提供描述信息。
  3. 上下文提供:此时,Host 应用可以将资源和提示提供给用户,或者将工具解析为 LLM 兼容的格式(例如,JSON 函数调用)。
  4. 调用:如果 LLM 判断需要使用某个工具(例如,基于用户请求“‘X’ 仓库中有哪些未解决的问题?”),Host 会指导客户端向适当的服务器发送调用请求。
  5. 执行:服务器收到请求(例如,fetch_github_issues,指定 repo 为 ‘X’),执行底层逻辑(调用 GitHub API),并获得结果。
  6. 响应:服务器将结果返回给客户端。
  7. 完成:客户端将结果传递回 Host,Host 将其整合到 LLM 的上下文中,允许 LLM 基于这些最新的外部信息生成最终的用户响应。

MCP 服务器

MCP 服务器是连接 MCP 世界和外部系统(如 API、数据库、本地文件等)功能的桥梁或 API。本质上,它们是外部能力的封装器,按照 MCP 规范公开这些能力。

MCP 服务器可以使用多种语言(如 Python、TypeScript、Java、Rust 等)构建,只要它们能够通过支持的传输方式进行通信。服务器主要通过两种方法与客户端通信:

  • stdio(标准输入/输出):当客户端和服务器运行在同一台机器上时使用。这对于本地集成(例如,访问本地文件或运行本地脚本)非常简单有效。
  • HTTP 通过 SSE(服务器发送事件):客户端通过 HTTP 连接到服务器。在初始设置后,服务器可以通过持久连接使用 SSE 标准向客户端推送消息(事件)。

使用 Python 和FastMCP构建 MCP 服务器的示例:

from fastmcp import FastMCP # Create an MCP server mcp = FastMCP("Demo") # Add a tool, will be converted into JSON spec for function calling @mcp.tool() def add(a: int, b: int) -> int: """Add two numbers""" return a + b # Add a data resource, e.g. displayed on new chats @mcp.resource("greeting://{name}") def get_greeting(name: str) -> str: """Get a personalized greeting""" return f"Hello, {name}!" # Specific prompt templates for better use @mcp.prompt() def review_code(code: str) -> str: return f"Please review this code:\n\n{code}"

预建和社区构建的 MCP 服务器列表:

https://github.com/punkpeye/awesome-mcp-servers

https://github.com/modelcontextprotocol/servers

https://mcp.composio.dev/

MCP 客户端

MCP 客户端是主机应用程序(IDE、聊天机器人等)的一部分,用于管理与特定 MCP 服务器的通信。

  • 角色:根据 MCP 规范处理连接管理、功能发现、请求转发和响应处理。
  • 主人/客户的示例:
    • UI 应用程序:Claude Desktop、Microsoft Copilot Studio、LibreChat、Claude Code
    • IDE:Cursor、Windsurf、Continue、Zed、Cline
    • 自定义代理(Python / TypeScript):
      • Firebase Genkit
      • 语言图谱
      • OpenAI 代理 SDK
      • ….

如何使用 Python 和 mcp 构建 MCP 客户端的示例。

from mcp import ClientSession, StdioServerParameters, types from mcp.client.stdio import stdio_client # Commands for running/connecting to MCP Server server_params = StdioServerParameters(
    command="python", # Executable args=["example_server.py"], # Optional command line arguments env=None, # Optional environment variables ) async with stdio_client(server_params) as (read, write): async with ClientSession(
        read, write, sampling_callback=handle_sampling_message
    ) as session: # Initialize the connection await session.initialize() # List available prompts prompts = await session.list_prompts() # Get a prompt prompt = await session.get_prompt( "example-prompt", arguments={"arg1": "value"}
        ) # List available resources resources = await session.list_resources() # List available tools tools = await session.list_tools() # Read a resource content, mime_type = await session.read_resource("file://some/path") # Call a tool result = await session.call_tool("tool-name", arguments={"arg1": "value"})
 

为什么会有这么大的炒作?MCP 赢了吗?

虽然 Anthropic 在 2024 年底宣布了 MCP,但其发展势头在 2025 年初显著加速。这不仅仅是随机炒作;有几个因素汇聚在一起:

  • “AI-Native”虽然存在用于 API 交互的旧标准(例如 OpenAPI、GraphQL 或 SOAP),但 MCP 是专门为满足现代 AI 代理的需求而设计的。MCP 改进了代理开发中看到的模式:
    • 工具(模型控制):人工智能决定采取的行动。
    • 资源(应用程序控制):提供给 AI 的上下文。
    • 提示(用户控制):特定的用户调用的交互。
  • 拥有强大后盾的“开放标准”:任何“开放标准”都应该有规范,而MCP 的规范非常好。仅凭这个规范就击败了许多没有提供如此详细规范的竞争对手。
  • 建立在经过验证的基础之上: Anthropic 不是从头开始重新发明一切,而是改编自语言服务器协议 (LSP),例如JSON-RPC 2.0
  • 强大的初始生态系统和 Dogfood 测试: MCP 并非以一个规范的形式推出。Anthropic 对其进行了广泛的“dogfood 测试”,并发布了一套全面的初始套件:
    • 客户:Claude Desktop。
    • 服务器:大量参考实现(文件系统、git、Slack 等)。
    • 工具:MCP Inspector 用于测试,有出色的文档
    • SDK:Python 和 TypeScript 库,现在还有 Java、Kotlin C#
  • 网络效应:开放性催生了一个社区。Cursor 和 Windsurf 等工具集成了 MCP。Composio 等公司为数百个集成提供了预构建的服务器。OpenAI 宣布支持 MCP。开发人员构建了数千个社区 MCP 服务器(GitHub、Slack、数据库、Docker 等)。

使用 Gemini 和 Python uSDK 的实际示例

from typing import List from google import genai from google.genai import types from mcp import ClientSession, StdioServerParameters from mcp.client.stdio import stdio_client import os
 
client = genai.Client(api_key=os.getenv("GEMINI_API_KEY"))
model = "gemini-2.0-flash" # Create server parameters for stdio connection server_params = StdioServerParameters(
    command="npx", # Executable args=[ "-y", "@openbnb/mcp-server-airbnb",
    ], # Optional command line arguments env=None, # Optional environment variables ) async def agent_loop(prompt: str, client: genai.Client, session: ClientSession): contents = [types.Content(role="user", parts=[types.Part(text=prompt)])] # Initialize the connection await session.initialize() # --- 1. Get Tools from Session and convert to Gemini Tool objects --- mcp_tools = await session.list_tools()
    tools = types.Tool(function_declarations=[
        { "name": tool.name, "description": tool.description, "parameters": tool.inputSchema,
        } for tool in mcp_tools.tools
    ]) # --- 2. Initial Request with user prompt and function declarations --- response = await client.aio.models.generate_content(
        model=model, # Or your preferred model supporting function calling contents=contents,
        config=types.GenerateContentConfig(
            temperature=0,
            tools=[tools],
        ), # Example other config ) # --- 3. Append initial response to contents --- contents.append(response.candidates[0].content) # --- 4. Tool Calling Loop ---  turn_count = 0 max_tool_turns = 5 while response.function_calls and turn_count < max_tool_turns: turn_count tool_response_parts: listtypes.part='[]' --- 4.1 process all function calls in order and return in this turn --- for fc_part in response.function_calls: tool_name='fc_part.name' args='fc_part.args' or ensure args is a dict printfattempting to call mcp tool: tool_name with args: args tool_response: dict try: call the sessions tool executor tool_result='await' session.call_tooltool_name args printfmcp tool tool_name executed successfully. if tool_result.iserror: tool_response='{"error":' tool_result.content0.text else: tool_response='{"result":' tool_result.content0.text except exception as e: tool_response='{"error":' ftool execution failed: typee.__name__: e prepare functionresponse part tool_response_parts.append types.part.from_function_response name='tool_name,' response='tool_response' --- 4.2 add the tool responses to history --- contents.appendtypes.contentrole='user' parts='tool_response_parts))' printfadded lentool_response_parts tool response parts to history. --- 4.3 make the next call to the model with updated history --- printmaking subsequent api call with tool responses... response='await' client.aio.models.generate_content model='model,' contents='contents,' send updated history config='types.GenerateContentConfig(' temperature='1.0,' tools='[tools],' keep sending same config contents.appendresponse.candidates0.content if turn_count>= max_tool_turns and response.function_calls:
        print(f"Maximum tool turns ({max_tool_turns}) reached. Exiting loop.")
 
    print("MCP tool calling loop finished. Returning final response.") # --- 5. Return Final Response --- return response async def run(): async with stdio_client(server_params) as (read, write): async with ClientSession(
            read,
            write,
        ) as session: # Test prompt prompt = "I want to book an apartment in Paris for 2 nights. 03/28 - 03/30" print(f"Running agent loop with prompt: {prompt}") # Run agent loop res = await agent_loop(prompt, client, session) return res
res = await run()
print(res.text)

如果保证安全、更新和身份验证呢?

MCP 是一种动态协议。该规范在 GitHub 上得到积极维护,上次更新于 03/26,提高了安全性、可扩展性和可用性。

  • 身份验证和安全(OAuth 2.1):该协议现在要求使用 OAuth 2.1 框架来对远程 HTTP 服务器进行身份验证
  • 改进的传输和效率:以前的 HTTP + SSE 传输将被更灵活的 Streamable HTTP 传输和对 JSON-RPC 批处理的支持所取代。
  • 更丰富的上下文和控制:新的工具注释提供了有关工具行为的更多元数据(例如,只读与破坏性)

致谢

本概述从几个优秀资源中汲取了灵感和信息,包括:

MCP 是什么?为什么突然间每个人都在谈论它?

https://huggingface.co/blog/Kseniase/mcp

什么是 MCP

https://python.useinstructor.com/blog/2025/03/27/understanding-model-context-protocol-mcp/#conclusion

我授予 Claude 对我的服务器的 root 访问权限...模型上下文协议解释

https://www.youtube.com/watch?v=HyzlYwjoXOQ

MCP 为何获胜

https://www.latent.space/p/why-mcp-won

使用模型上下文协议构建代理 - 与 Anthropic 的 Mahesh Murag

https://www.youtube.com/watch?v=kQmXtrmQ5Zg

立即下载

相似工具

评论列表 共有 0 条评论

暂无评论
发表
评论
顶部