MCP Server Tool 开发学习文档

MCP Server Tool 简介
核心开发流程与知识点详解
- 2.1 工具函数的实现
- 2.2 MCP Server 的注册与启动
- 2.3 工具注册与调用机制
- 2.4 工具列表的声明与返回
- 2.5 传输方式（stdio 与 sse）
Python 源码详细解析
SSE 方式本地部署方法
客户端访问与调用示例

1. MCP Server Tool 简介

MCP（Model Context Protocol）是一种用于模型与外部工具交互的协议。MCP Server Tool 是基于 MCP 协议开发的服务端工具，能够通过标准输入输出（stdio）或服务端事件（SSE）等方式暴露自定义工具，供客户端远程调用。

本例实现了一个简单的“网站抓取”工具，支持通过 MCP 协议获取指定网页内容。

2. 核心开发流程与知识点详解

2.1 工具函数的实现

工具函数是 MCP Server Tool 的核心业务逻辑。例如本例中的 fetch_website，用于异步抓取网页内容：

async def fetch_website(url: str) -> list[types.TextContent | types.ImageContent | types.EmbeddedResource]:headers = {"User-Agent": "MCP Test Server (github.com/modelcontextprotocol/python-sdk)"}async with create_mcp_http_client(headers=headers) as client:response = await client.get(url)response.raise_for_status()return [types.TextContent(type="text", text=response.text)]

异步实现：利用 async/await，适合高并发场景。
类型注解：返回值为内容对象列表，支持文本、图片、嵌入资源。
自定义 User-Agent：便于调试和识别请求来源。

2.2 MCP Server 的注册与启动

MCP Server 通过 Server 类实例化，并注册工具与工具列表：

app = Server("mcp-website-fetcher")

Server 名称：用于标识服务实例。
注册工具与工具列表：通过装饰器 @app.call_tool() 和 @app.list_tools() 实现。

2.3 工具注册与调用机制

工具注册函数用于处理客户端的工具调用请求：

@app.call_tool()
async def fetch_tool(name: str, arguments: dict) -> list[types.TextContent | types.ImageContent | types.EmbeddedResource]:if name != "fetch":raise ValueError(f"Unknown tool: {name}")if "url" not in arguments:raise ValueError("Missing required argument 'url'")return await fetch_website(arguments["url"])

参数校验：确保工具名和参数正确。
调用业务逻辑：将参数传递给实际的工具函数。

2.4 工具列表的声明与返回

通过 @app.list_tools() 注册工具列表，供客户端查询：

@app.list_tools()
async def list_tools() -> list[types.Tool]:return [types.Tool(name="fetch",description="Fetches a website and returns its content",inputSchema={"type": "object","required": ["url"],"properties": {"url": {"type": "string","description": "URL to fetch",}},},)]

Tool 对象：描述工具名称、功能、输入参数结构。
inputSchema：采用 JSON Schema 格式，便于前端自动生成表单或校验参数。

2.5 传输方式（stdio 与 sse）

MCP Server 支持两种传输方式：

stdio：通过标准输入输出进行通信，适合本地或嵌入式场景。
sse：通过 HTTP SSE（Server-Sent Events）协议，适合 Web 场景。

切换方式通过命令行参数 --transport 控制：

@click.option("--transport", type=click.Choice(["stdio", "sse"]), default="stdio", help="Transport type")

stdio 模式下，使用 mcp.server.stdio.stdio_server 启动服务。
sse 模式下，使用 starlette 框架和 uvicorn 启动 HTTP 服务，并挂载 SSE 路由。

3. Python 源码详细解析

3.1 入口函数

@click.command()
@click.option("--port", default=8000, help="Port to listen on for SSE")
@click.option("--transport", type=click.Choice(["stdio", "sse"]), default="stdio", help="Transport type")
def main(port: int, transport: str) -> int:...

使用 click 实现命令行参数解析。
根据 transport 参数选择不同的服务启动方式。

3.2 stdio 模式

from mcp.server.stdio import stdio_serverasync def arun():async with stdio_server() as streams:await app.run(streams[0], streams[1], app.create_initialization_options())anyio.run(arun)

通过 anyio 启动异步主循环。
适合本地开发和调试。

3.3 sse 模式

from mcp.server.sse import SseServerTransport
from starlette.applications import Starlette
from starlette.responses import Response
from starlette.routing import Mount, Routesse = SseServerTransport("/messages/")async def handle_sse(request):async with sse.connect_sse(request.scope, request.receive, request._send) as streams:await app.run(streams[0], streams[1], app.create_initialization_options())return Response()starlette_app = Starlette(debug=True,routes=[Route("/sse", endpoint=handle_sse, methods=["GET"]),Mount("/messages/", app=sse.handle_post_message),],
)import uvicorn
uvicorn.run(starlette_app, host="0.0.0.0", port=port)

使用 starlette 框架搭建 HTTP 服务。
/sse 路由用于建立 SSE 连接。
/messages/ 路由用于消息传递。

4. SSE 方式本地部署方法

4.1 安装依赖

确保已安装 uvicorn、starlette、mcp 相关依赖：

pip install uvicorn starlette mcp

4.2 启动服务

在 python-sdk/examples/servers/simple-tool/ 目录下运行：

uv run mcp-simple-tool --transport sse --port 8000

--transport sse：指定使用 SSE 传输方式。
--port 8000：指定监听端口（可自定义）。

4.3 服务启动后，SSE 端点为：

/sse：用于建立 SSE 连接
/messages/：用于消息通信

5. 客户端访问与调用示例

5.1 客户端代码示例（STDIO，可类比改为 SSE）

import asyncio
from mcp.client.session import ClientSession
from mcp.client.sse import SseServerParameters, sse_clientasync def main():async with sse_client(SseServerParameters(url="http://localhost:8000/sse")) as (read, write):async with ClientSession(read, write) as session:await session.initialize()# 列出可用工具tools = await session.list_tools()print(tools)# 调用 fetch 工具result = await session.call_tool("fetch", {"url": "https://example.com"})print(result)asyncio.run(main())