什么是 MCP

“MCP 服务”指的是基于 模型上下文协议 (Model Context Protocol) 构建的服务。MCP 是一种新兴的开放协议，旨在为大型语言模型 (LLMs) 和其他 AI 应用提供一种标准化的方式来安全、高效地访问外部数据、工具和服务，从而增强它们的能力，超越其训练数据的限制。

你可以将 MCP 服务理解为 AI 应用的“USB-C”接口。它定义了一套通用的规则和格式，使得 AI 模型可以通过一个统一的方式与各种外部资源进行交互，而无需为每个不同的工具编写特定的集成代码。

举个例子：

有一天觉得自己的头发太长了。于是你就跟自己的 AI 小助手说：“你想去隔壁村帅气美发店，弄个靓仔发型。”

这时，你的 AI 助手需要完成一系列操作：

• 查看你的手机日程：确认你今天下午3点后有空
• 调用高德地图：查询去"帅气美发店"的路线和预计用时
• 访问大众点评：查看店铺评价、价格，并帮你线上排队

如果没有 MCP 协议，这个过程会很麻烦：

• 每个 App 都有不同的接口，AI 要分别学习对接
• 你要反复授权，还可能泄露隐私数据
• AI 可能记错信息，比如把预约时间搞混

有了 MCP 协议后：

• AI 通过标准化的 MCP 接口，像"插U盘"一样快速接入你的日历、地图和点评数据
• 自动保持上下文：知道"弄发型"关联到"查路线+预约"
• 安全可控：只获取必要权限（比如只看今天日程，不翻旧记录）

最后，AI 一气呵成地回复：“已预约帅气美发店下午3:30（评分4.8星），打车15分钟可达。要现在叫车吗？”

MCP 的核心概念

在 Anthropic 自己的官网上，有 MCP 的基础介绍，最核心的概念包括：

• MCP Host（如 Claude Desktop）：运行 AI 应用的主程序。
• MCP Client：连接服务器的协议客户端。
• MCP Server：轻量级服务，提供特定功能（如文件操作、数据库查询）。
• 数据源：本地（如 SQLite）或远程（如 AWS API）

从这个架构上看得出 MCP Server 是跑在本地的小型服务，但实际上，目前已经支持远程 MCP 服务了，并可以通过 Anthropic 提供的 SDK 挂载到各类服务器上。

MCP Server 核心能力

MCP 服务器可提供三种核心能力类型：

1. 资源 (Resources)： 客户端可读取的类文件数据（例如 API 响应或文件内容）
2. 工具 (Tools)： 大型语言模型可调用的功能函数（需用户批准执行）
3. 提示 (Prompts)： 预置的任务模板，辅助用户完成特定操作流程

Anthropic 有提供官方的各类 SDK，我以 Python SDK 中的文档为例，做内容翻译和整理。

资源（Resources）

用于向 LLM 提供数据，类似 REST API 的 GET 端点：

• 无副作用：仅暴露数据，不执行复杂计算或修改操作。
• 静态与动态资源：支持固定配置（如 config://app）和动态参数化路径（如 users://{user_id}/profile）。

示例代码：

@mcp.resource("users://{user_id}/profile")
def get_user_profile(user_id: str) -> str:
    return f"Profile for {user_id}"

工具（Tools）

允许 LLM 通过服务器执行操作，可包含副作用：

• 同步与异步支持：支持普通函数和异步函数（如 HTTP 请求）。
• 参数强类型化：通过函数参数定义工具输入，返回值自动序列化为 JSON。

示例代码：

@mcp.tool()
async def fetch_weather(city: str) -> str:
    async with httpx.AsyncClient() as client:
        return await client.get(f"https://api.weather.com/{city}")

提示（Prompts）

可复用的交互模板，用于结构化 LLM 交互：

• 灵活输出格式：可返回纯文本或消息对象列表（如 UserMessage、AssistantMessage）。

示例代码：

@mcp.prompt()
def debug_error(error: str) -> list[base.Message]:
    return [
        base.UserMessage("错误信息："),
        base.UserMessage(error),
        base.AssistantMessage("请描述已尝试的解决方法。")
    ]

MCP Server 的传入参数

图像处理（Images）

通过 Image 类简化图像数据传输：

• 自动格式转换：支持从 PIL 图像对象转换为字节数据，并指定格式（如 PNG）。

示例代码：

@mcp.tool()
def create_thumbnail(image_path: str) -> Image:
    img = PILImage.open(image_path)
    return Image(data=img.tobytes(), format="png")

上下文（Context）

为工具和资源提供运行时能力：

• 核心功能：

  • 进度报告（ctx.report_progress()）
  • 日志记录（ctx.info()）
  • 资源读取（ctx.read_resource()）

示例代码：

@mcp.tool()
async def long_task(files: list[str], ctx: Context):
    for file in files:
        await ctx.read_resource(f"file://{file}")

认证（Authentication）

基于 OAuth 2.0 的安全访问控制：

• 配置项：

  • 颁发者 URL（issuer_url）
  • 客户端注册范围（valid_scopes）
  • 强制权限（required_scopes）

示例代码：

mcp = FastMCP("My App", 
    auth_server_provider=MyOAuthServerProvider(),
    auth=AuthSettings(required_scopes=["myscope"])
)

一个简单的例子（Python、SSE）

这里实现一个 MCP Server 和一个 MCP Client。
我参考 官方文档了，这里做简化、整理和翻译。
另外官方的例子是基于 stdin 来实现的，即通过本地来完成通信的。考虑到 MCP 也支持 SSE 连接，所以这里的例子使用 SSE 示例。

安装必要的库

pip install mcp[cli] httpx asgiref uvicorn starlette

Server 部分

# mcp_server.py
# mcp_server.py
from typing import Any
import httpx
from mcp.server.fastmcp import FastMCP
from starlette.applications import Starlette
from starlette.routing import Mount
import uvicorn

# 初始化 mcp 服务
mcp1 = FastMCP("test helper")

# 添加一个资源
@mcp1.resource("greeting://{name}")
def get_greeting(name: str) -> str:
    """Get a personalized greeting"""
    return f"Hello, {name}!"

@mcp1.resource("info://")
def get_info() -> str:
    """Get a helper info"""
    return f"My name is Wangwei"

# 添加一个工具
@mcp1.tool()
def add(a: int, b: int) -> int:
    """Add two numbers"""
    return a + b

# 添加一个 prompt
@mcp1.prompt()
def gen_sum(content: str) -> str:
    """generate summary"""
    return f"请根据 content 的内容生成一段总结，不超过100个字符。\ncontent: {content}\n总结:"


if __name__ == "__main__":
    # 初始化服务
    app = Starlette(
    routes=[
            Mount('/', app=mcp1.sse_app()), # 服务会跑在 /sse 下
        ]
    )
    # 运行
    uvicorn.run(app, host="0.0.0.0", port=8855)

Client 部分

# mcp_client.py
from mcp.client.sse import sse_client
from mcp import ClientSession


async def main():
    # Connect to a streamable HTTP server
    async with sse_client("http://localhost:8855/sse") as (
        read_stream,
        write_stream
    ):
        # 创建一个 session
        async with ClientSession(read_stream, write_stream) as session:
            # 初始化连接
            await session.initialize()

            # 查看有哪些 resources
            resources = await session.list_resources()
            print('===资源列表===')
            for r in resources.resources:
                print(r)
            # 查看有哪些 tools
            tools = await session.list_tools()
            print('===工具列表===')
            for t in tools.tools:
                print(t)
            # 查看有哪些 prompts
            prompts = await session.list_prompts()
            print('===prompt列表===')
            for p in prompts.prompts:
                print(p)

            print('\n\n')

            # 调用资源
            print('===调用资源===')
            greeting = await session.read_resource("greeting://world")
            print(greeting)
            
            # 调用工具
            print('===调用工具===')
            tool_result = await session.call_tool("add", {"a": 1, "b": 2})
            print(tool_result)

            # 调用 prompt
            print('===调用 prompt===')
            prompt_result = await session.get_prompt("gen_sum", {"content": "我是一段超过 1000 字符的文章"})
            print(prompt_result)

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

命令行测试运行

# 开一个终端
python mcp_server.py
# 另开一个终端
python mcp_client.py

你能看到这样的结果

===资源列表===
uri=Url('info://') name='get_info' description='Get a helper info' mimeType='text/plain' size=None annotations=None
===工具列表===
name='add' description='Add two numbers' inputSchema={'properties': {'a': {'title': 'A', 'type': 'integer'}, 'b': {'title': 'B', 'type': 'integer'}}, 'required': ['a', 'b'], 'title': 'addArguments', 'type': 'object'} annotations=None
===prompt列表===
name='gen_sum' description='generate summary' arguments=[PromptArgument(name='content', description=None, required=True)]



===调用资源===
meta=None contents=[TextResourceContents(uri=Url('greeting://world'), mimeType='text/plain', text='Hello, world!')]
===调用工具===
meta=None content=[TextContent(type='text', text='3', annotations=None)] isError=False
===调用 prompt===
meta=None description=None messages=[PromptMessage(role='user', content=TextContent(type='text', text='请根据 content 的内容生成一段总结，不超过100个字符。\ncontent: 我是一段超过 1000 字符的文章\n总结:', annotations=None))]

目前看，带有参数的 resources 不会被 list_resources 列出。

用 MCP Host 测试

MCP Host 已经有很多，这个项目中也列出了不少。
我自己在用 Trae，所以在 Trae 中进行 MCP 的配置：

{
  "mcpServers": {
    "test": {
      "url": "http://127.0.0.1:8855/sse"
    }
  }
}

配置如图

配置后

回到对话框界面，输入“帮我计算 2+3 等于几”，可以看到 MCP 服务已经加载

执行后获得结果