1. 引言

随着大语言模型能力的持续增强,AI Agent 正在从「能聊天」走向「能干活」。然而,要让 Agent 真正调用外部工具、访问业务数据,开发者面临一个核心难题:如何让模型与工具之间高效、标准化地通信?MCP(Model Context Protocol,模型上下文协议)正是为解决这一问题而生的开放协议。

本文将带你从零开始,理解 MCP 的核心概念,动手实现一个完整的 MCP Server,并将其接入 AI Agent,最终搭建出一条可复用的工具链。无论你是后端工程师、AI 应用开发者,还是对 Agent 架构感兴趣的爱好者,本文都能为你提供一条清晰、可落地的实战路径。

2. MCP 协议核心概念

在动手写代码之前,先建立对 MCP 的整体认知。本节介绍 MCP 是什么、解决什么问题,以及它的核心架构与通信模型。

2.1 什么是 MCP

MCP 是由 Anthropic 于 2024 年底提出的开放协议,旨在为 AI 应用(Host)与外部工具/数据源(Server)之间建立标准化的连接方式。你可以把它理解为「AI 世界的 USB-C 接口」:只要设备支持这个标准,就能即插即用。

2.2 MCP 解决的核心问题

在没有 MCP 之前,每个 Agent 接入一个工具,都需要定制一套 API 封装、鉴权逻辑和调用协议,重复造轮子且难以复用。MCP 通过统一协议,将「工具定义、调用、结果返回」标准化,让工具开发者只需实现一次 Server,即可被任意支持 MCP 的 Host 复用。

2.3 核心架构与角色

MCP 架构中包含三个关键角色:

  • Host:AI 应用本身,如 Claude Desktop、自研 Agent,负责与用户交互并调度模型。
  • Client:Host 内部的连接组件,负责与 Server 建立会话、收发消息。
  • Server:工具/数据源的提供方,暴露标准化的工具、资源和提示词。

三者关系如下图所示:

JSON-RPC 2.0

Host(AI 应用)

Client

MCP Server

工具 / 数据源

2.4 通信模型与消息格式

MCP 基于 JSON-RPC 2.0 进行通信,支持两种传输方式:

  • stdio:Server 作为子进程启动,通过标准输入输出通信,适合本地开发。
  • HTTP + SSE:Server 作为独立服务部署,通过 HTTP 长连接通信,适合生产环境。

每条消息包含 jsonrpcmethodparamsid 等字段,协议定义了初始化、工具列表、工具调用等标准方法。

3. 环境准备与项目初始化

本节搭建开发环境,创建项目骨架,为后续编码做好准备。

3.1 开发环境要求

  • Python 3.10+ 或 Node.js 18+
  • 一个支持 MCP 的 Host(如 Claude Desktop,或使用官方 MCP Inspector 调试)
  • 包管理工具(pip / npm)

3.2 初始化项目

以 Python 为例,创建项目目录并安装官方 SDK:

mkdir mcp-agent-toolkit
cd mcp-agent-toolkit
python -m venv .venv
source .venv/bin/activate
pip install mcp

3.3 验证 SDK 安装

python -c "import mcp; print(mcp.__version__)"

4. 实现第一个 MCP Server

本节从零实现一个功能完整的 MCP Server,包含工具定义、注册与运行。

4.1 定义工具

我们实现一个「获取天气」的工具,演示工具定义、参数校验与结果返回:

from mcp.server import Server
from mcp.server.stdio import stdio_server
from mcp.types import Tool, TextContent

app = Server("weather-server")

@app.list_tools()
async def list_tools():
    return [
        Tool(
            name="get_weather",
            description="查询指定城市的实时天气",
            inputSchema={
                "type": "object",
                "properties": {
                    "city": {"type": "string", "description": "城市名称"}
                },
                "required": ["city"]
            }
        )
    ]

@app.call_tool()
async def call_tool(name: str, arguments: dict):
    if name == "get_weather":
        city = arguments["city"]
        # 这里替换为真实天气 API 调用
        result = f"{city} 今日晴,25°C,微风"
        return [TextContent(type="text", text=result)]
    raise ValueError(f"未知工具: {name}")

4.2 启动 Server

async def main():
    async with stdio_server() as (read_stream, write_stream):
        await app.run(read_stream, write_stream)

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

4.3 用 MCP Inspector 调试

MCP 官方提供了可视化调试工具 Inspector,可以快速验证 Server 的工具列表与调用结果:

mcp dev server.py

5. 将 Server 接入 AI Agent

Server 就绪后,本节演示如何将其接入一个真实的 AI Agent,让模型能够自主调用工具。

5.1 配置 Host 连接

以 Claude Desktop 为例,在配置文件中声明 MCP Server:

{
  "mcpServers": {
    "weather": {
      "command": "python",
      "args": ["server.py"],
      "cwd": "/path/to/mcp-agent-toolkit"
    }
  }
}

5.2 在 Agent 中调用工具

在自研 Agent 中,通过 SDK 连接 Server 并获取工具列表:

from mcp.client.stdio import stdio_client
from mcp.client.session import ClientSession

async def query_tool():
    async with stdio_client(["python", "server.py"]) as (read, write):
        async with ClientSession(read, write) as session:
            await session.initialize()
            tools = await session.list_tools()
            result = await session.call_tool("get_weather", {"city": "北京"})
            print(result)

5.3 完整工具链流程

用户提问

Agent 理解意图

需要工具?

MCP Client 调用 Server

工具执行并返回结果

Agent 组织回答

返回用户

6. 进阶:构建多工具组合与状态管理

单个工具只是起点,真实场景往往需要多个工具协作。本节介绍如何扩展 Server、管理工具间状态,并处理错误。

6.1 注册多个工具

list_tools 中返回多个 Tool 对象,并在 call_tool 中按 name 分发即可。建议将每个工具封装为独立函数,便于维护与测试。

6.2 工具间共享状态

当工具需要共享数据(如用户会话、缓存)时,可在 Server 内部维护一个状态对象:

class ToolContext:
    def __init__(self):
        self.cache = {}

context = ToolContext()

@app.call_tool()
async def call_tool(name: str, arguments: dict):
    if name == "save_note":
        context.cache[arguments["key"]] = arguments["value"]
        return [TextContent(type="text", text="已保存")]
    if name == "get_note":
        value = context.cache.get(arguments["key"], "未找到")
        return [TextContent(type="text", text=value)]

6.3 错误处理与超时

工具调用可能失败或超时,Server 应返回结构化错误信息,Host 侧也应设置合理的超时与重试策略:

@app.call_tool()
async def call_tool(name: str, arguments: dict):
    try:
        # 业务逻辑
        pass
    except Exception as e:
        return [TextContent(type="text", text=f"工具执行失败: {str(e)}")]

7. 生产环境部署与安全

从本地原型走向生产,需要解决部署形态、鉴权、审计等工程问题。

7.1 部署形态选择

  • 本地 stdio:适合个人工具、开发调试。
  • 远程 HTTP + SSE:适合团队共享、服务化部署,需配合网关做鉴权与限流。

7.2 安全最佳实践

  • 所有工具调用必须经过鉴权,禁止 Server 暴露敏感操作。
  • 对工具入参做严格校验,防止注入攻击。
  • 记录完整调用日志,便于审计与排障。
  • 对第三方 API 调用设置超时与熔断。

7.3 可观测性

为 Server 接入日志、指标与链路追踪,便于定位问题:

import logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger("mcp-server")

@app.call_tool()
async def call_tool(name: str, arguments: dict):
    logger.info("调用工具 %s, 参数 %s", name, arguments)
    # ...

8. 总结与展望

本文从 MCP 的核心概念出发,完整走通了「定义工具 → 实现 Server → 接入 Agent → 多工具协作 → 生产部署」的实战链路。MCP 的价值在于标准化:一次实现,处处复用,让 AI Agent 的工具生态得以快速生长。

未来,MCP 生态会持续演进,工具市场、跨组织共享、更丰富的资源类型都将成为现实。建议你从一个小工具开始,亲手跑通这条链路,再逐步扩展自己的 Agent 工具链。

Logo

欢迎加入 MCP 技术社区!与志同道合者携手前行,一同解锁 MCP 技术的无限可能!

更多推荐