Model Context Protocol (MCP)

模型上下文协议（MCP）

模型上下文协议（MCP）是一种旨在为大模型（LLM）与外部工具或数据源建立通信通道的协议。它采用客户端-服务端架构，并使用 JSON-RPC 2.0 进行消息传递。 MCP 是 Anthropic 几个月前才推出的，但很快就在 AI 社区中受到了广泛关注。 在下面的章节中，我将介绍 MCP 规范，并提供一些 Python 示例。

---

MCP 服务端

MCP 服务端为客户端提供三项关键能力：

能力	描述
工具	执行逻辑（例如计算器）、触发副作用（例如修改环境）或受 CPU 限制的函数或服务。
资源	任意类型的数据，例如图像、文件等。
提示词	客户端可以使用的提示词模板。

MCP 服务端可以使用 FastMCP 或低级 API 实现。 它支持通过 HTTP 上的 SSE（服务器发送事件）或 stdin 进行通信。 本地通信使用 stdin，基于网络的通信使用 SSE。 这种通信类型在 MCP 规范中称为传输层。

FastMCP

假设我们要为基于大模型的英语教学应用实现一个 MCP 服务端。 此 MCP 服务端将提供一份测验（资源）、一个用于确定学习者水平的分级函数（工具），以及一个提示词模板（提示词）。

from mcp.server.fastmcp import FastMCP

# You can customize important parameters by passing them to FastMCP.
# In this example, all key parameters are set to their default values,
# but you can modify them according to your needs.
# sse_url = http://0.0.0.0:8000/sse
mcp = FastMCP(
    name="MCPServer",
    debug=True,
    host="0.0.0.0",
    port=8000,
    sse_path="/sse",
    message_path="/messages/",
    log_level="DEBUG",
)


@mcp.resource(
    uri="https://quiz.xyz",
    name="GetQuiz",
    description="Provides a link to an online English level assessment quiz.",
)
def get_quiz() -> str:
    return "Link to online quiz: https://quiz.xyz"


@mcp.tool(
    name="FindLevel",
    description="Determines the student's English level based on their quiz score.",
)
def find_level(grade: int) -> str:
    if grade < 50:
        return "Beginner"
    if grade < 75:
        return "Intermediate"
    return "Expert"


@mcp.prompt(
    name="GetPrompt",
    description="Generates a prompt to ask an LLM to teach English based on the student's level.",
)
def get_prompt(name: str, level: str) -> str:
    return f"Teach {name} English based on this level: {level}."

通过 SSE 提供 FastMCP 服务有多种方式：sse_app、run_sse_async 或 run。 第一种方式使用 ASGI 框架（例如 Starlette）。 第二种方式利用 Python 的 asyncio。 第三种方式是 mcp 实例的内置函数，其底层使用 run_sse_async。

from starlette.applications import Starlette
from starlette.routing import Mount
import uvicorn


if __name__ == "__main__":
    app = Starlette(debug=True, routes=[
        Mount("/", mcp.sse_app()),
    ])
    uvicorn.run(app)

import asyncio

async def main():
    await mcp.run_sse_async()


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

# Pass transport=stdio for stdio server
if __name__ == "__main__":
    mcp.run(transport="sse")

低级 API

使用低级 API 实现服务端需要花费较多时间。 你必须实例化 Server 类，并实现一系列方法来列出和调用所有能力。 mcp python-sdk 的官方仓库此处提供了关于如何实现的详细指南。

---

MCP 客户端

根据服务端的暴露方式，有两种实现客户端的方式：sdio-client 和 sse-client。

SSE 客户端

在 fastmcp-server.py 中将传输方式设为 sse。

import asyncio
import logging
from typing import Union
from urllib.parse import urlparse

from anyio.streams.memory import MemoryObjectReceiveStream, MemoryObjectSendStream

import mcp.types as types
from mcp.client.session import ClientSession
from mcp.client.sse import sse_client
from mcp.shared.session import RequestResponder

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger("client")


async def message_handler(
    message: Union[
        RequestResponder[types.ServerRequest, types.ClientResult],
        types.ServerNotification,
        Exception,
    ],
) -> None:
    if isinstance(message, Exception):
        logger.error("Error: %s", message)
        return
    logger.info("Received message from server: %s", message)


async def run_session(
    read_stream: MemoryObjectReceiveStream,
    write_stream: MemoryObjectSendStream,
) -> None:
    async with ClientSession(
        read_stream,
        write_stream,
        message_handler=message_handler,
    ) as session:
        initialize_response = await session.initialize()
        logger.info(f"Initialized response: {initialize_response}")
        tools = await session.list_tools()
        tools = tools.tools
        logger.info(f"List of tools: {[tool.name for tool in tools]}")
        response = await session.call_tool("FindLevel", {"grade": 86})
        logger.info(f"Find level response: {response}")


async def main(command_or_url: str) -> None:
    if urlparse(command_or_url).scheme in ("http", "https"):
        async with sse_client(command_or_url) as streams:
            await run_session(*streams)


if __name__ == "__main__":
    sse_url = "http://0.0.0.0:8000/sse"  # change url
    asyncio.run(main(sse_url))

Stdio 客户端

你必须传入服务端脚本以运行 stdio 客户端（在 fastmcp-server.py 中将传输方式设为 stdio）：

python stdio.py fastmcp-server.py

import asyncio
import sys
import logging
from contextlib import AsyncExitStack
from typing import Optional

from mcp import ClientSession, StdioServerParameters
from mcp.client.stdio import stdio_client

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger("client")


class MCPClient:
    def __init__(self) -> None:
        self.session: Optional[ClientSession] = None
        self.exit_stack = AsyncExitStack()

    async def connect_to_server(self, server_script_path: str) -> None:
        if not server_script_path.endswith(".py"):
            raise ValueError("Server script must be a .py file")

        server_params = StdioServerParameters(
            command="python",
            args=[server_script_path],
            env=None,
        )

        stdio_transport = await self.exit_stack.enter_async_context(
            stdio_client(server_params)
        )
        self.stdio, self.write = stdio_transport

        self.session = await self.exit_stack.enter_async_context(
            ClientSession(self.stdio, self.write)
        )
        initialize_response = await self.session.initialize()
        logger.info(f"Initialized response: {initialize_response}")

    async def list_tools(self) -> None:
        if not self.session:
            raise RuntimeError("Session is not initialized.")
        response = await self.session.list_tools()
        tools = response.tools
        logger.info(f"List of tools: {[tool.name for tool in tools]}")

    async def find_level(self, grade: int) -> None:
        response = await self.session.call_tool("FindLevel", {"grade": grade})
        logger.info(f"Find level response: {response}")

    async def cleanup(self) -> None:
        await self.exit_stack.aclose()


async def main() -> None:
    if len(sys.argv) < 2:
        print("Usage: python client.py <path_to_server_script>")
        sys.exit(1)

    client = MCPClient()
    try:
        await client.connect_to_server(sys.argv[1])
        await client.list_tools()
        await client.find_level(86)
    except Exception as exc:
        print("Error:", exc)
    finally:
        await client.cleanup()


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

---

协议规范

客户端与服务端之间会交换三种类型的消息。 所有消息均使用 JSON-RPC 2.0 格式，以支持远程过程调用。

• 请求：需要响应的请求（JSONRPCRequest）。
• 响应：对请求的成功（非错误）响应（JSONRPCResponse）。
• 通知：不需要响应的通知（JSONRPCNotification）。

标准 JSON-RPC 错误代码：

• PARSE_ERROR = -32700
• INVALID_REQUEST = -32600
• METHOD_NOT_FOUND = -32601
• INVALID_PARAMS = -32602
• INTERNAL_ERROR = -32603

初始化

客户端应先与服务端建立成功连接，再发送消息。

1. 客户端发送初始化请求（InitializeRequest）。
2. 服务端回复其协议版本和功能（InitializeResult）。
3. 客户端发送已初始化通知以确认（InitializedNotification）。
4. 连接建立。

发现

客户端可以：

• 列出工具：ListToolsRequest/ ListToolsResult
• 列出资源：ListResourcesRequest/ ListResourcesResult
• 列出提示词：ListPromptsRequest/ ListPromptsResult

调用

客户端可以：

• 调用工具：CallToolRequest/ CallToolResult
• 获取提示词：GetPromptRequest/ GetPromptResult
• 读取资源：ReadResourceRequest /ReadResourceResult

---

Gradio + MCP

为你的 Gradio 应用提供 MCP 服务简单又顺畅，是我最喜欢的 MCP 集成之一。

1. 安装带有 MCP extra 的 Gradio：

   pip install "gradio[mcp]"
   

2. 使用 mcp_server=True 启动应用。
3. MCP 服务地址会显示在控制台中。你可以在应用页面底部查看 MCP 文档。

• 如需完整详情，请参阅 Gradio MCP 服务指南。
• 试试我在 Hugging Face 上的演示 Space。

---

参考资料

0. 介绍 Model Context Protocol
1. MCP Python SDK/ Github
2. Model Context Protocol/ Github
3. Model Context Protocol 文档
4. Hugging Face 上的 MCP 课程