如何开发[AI应用MCP] MCP ServerTool？
摘要：0 序 MCP := 模型上下文协议 := AI Agent应用与外部系统集成的标准协议 一言以蔽之，MCP = AI LLM(大语言模型) 与外部系统的集成协议 本文旨在总结和分享 MCP Server 和 MCP Tool 的开发经验。
0 序
MCP := 模型上下文协议 := AI Agent应用与外部系统集成的标准协议
一言以蔽之，MCP = AI LLM(大语言模型) 与外部系统的集成协议
本文旨在总结和分享 MCP Server 和 MCP Tool 的开发经验。
总结踩坑的日志。
旨在用最精简的方式剖析/展示 MCP 应用的开发过程细节。S
欢迎交流。
1 概述： MCP Server/Tool 开发指南
MCP 概念篇
[AIGC/Agent] MCP := 模型上下文协议 := AI Agent应用与外部系统集成的标准协议 - 博客园/千千寰宇 【推荐】
MCP 典型应用场景
文件系统访问：让 AI 安全地读写本地文件
数据库查询：SQL 执行、数据检索
API 集成：GitHub、Slack、自定义业务 API
代码执行：在沙箱中运行 Python/R/Java/... 代码
MCP 核心架构与关键技术栈
MCP 是一种开放协议，用于标准化 AI 模型与外部数据源、工具之间的连接方式。它采用 Client-Server 架构：
┌─────────────────┐ ┌─────────────────────────┐ ┌─────────────────┐
│ AI 应用/宿主 │ ←──→ │ MCP Client │ ←──→ │ MCP Server │
│ (Claude/Cursor)│ │ (Python/Java/... SDK) │ │ (数据源/工具) │
└─────────────────┘ └─────────────────────────┘ └─────────────────┘
↑
│ JSON-RPC 2.0 over stdio/SSE
↓
┌─────────────────────┐
│ MCP Server │
│ (文件系统/数据库/...)│
└─────────────────────┘
关键技术栈
SSE/Server-sent Events： 服务器推送事件到客户端；
适用于：需要实时数据推送的场景; 特点：单向通信（服务器 → 客户端），基于 HTTP 长连接
SSE(Server Send Events) ：服务器 => 浏览器的消息推送解决方案 - 博客园/千千寰宇
JSON-RPC 2.0
[JSON/RPC/MCP] JSON-RPC 2.0 : 轻量级远程过程调用协议 - 博客园/千千寰宇
ASGI application
[Web/Python] ASGI : 异步 Python Web 网关服务器 - 博客园/千千寰宇
2 实践案例
2.1 Python 原生版 MCP（Model Context Protocol）
2.1.0 安装 Python MCP
pip install mcp[cli]
2.1.1 Python MCP 运行流程(FastMCP版)
MCP SSE Server端
python mcp_sse_server.py
from mcp.server import FastMCP
app = FastMCP('know-data-mcp', port=18001)
@app.tool()
async def add(a: int, b: int) -> int:
"""加法运算"""
return a + b
@app.tool()
def minus(a: int, b: int) -> int:
"""Subtract two numbers"""
return a - b
@app.tool()
async def know_data() -> str:
"""
获取know-data的信息
Returns:
know-data的信息
"""
return "know-data是个喜欢捣鼓技术的软件工程师"
# 直接运行，就启用了1个api接口了，默认路径是/sse，可以在FastMCP中增加参数`sse_path`修改。
if __name__ == "__main__":
app.run(transport='sse')
FastMCP : "A more ergonomic interface for MCP servers."
启动后：
C:\Users\xxx\.conda\envs\ai-env\python.exe D:\Workspace\xxx\DemoPythonProject\mcp\mcp_server.py
INFO: Started server process [20736]
INFO: Waiting for application startup.
INFO: Application startup complete.
INFO: Uvicorn running on http://127.0.0.1:18001 (Press CTRL+C to quit)
MCP SSE Client端(curl版)
MCP客户端(窗口1): 向MCP Server发起请求，获得 sessionId
$ curl -N http://localhost:18001/sse
event: endpoint
data: /messages/?session_id=2dcb98b6eaa74e2eade0e07160af6610
: ping - 2026-03-25 16:56:39.972535+00:00
: ping - 2026-03-25 16:56:54.972487+00:00
MCP客户端(窗口2): 基于 sessionId 向MCP Server发起【会话初始化】请求
$ curl -X POST "http://127.0.0.1:18001/messages/?session_id=$SESSION_ID" \
-H "Content-Type: application/json" \
-d '{
"jsonrpc": "2.0",
"id": 1,
"method": "initialize",
"params": {
"protocolVersion": "2024-11-05",
"capabilities": {},
"clientInfo": { "name": "test-client", "version": "1.0.0" }
}
}'
Accepted
此时，MCP客户端(窗口1)将输出:
event: message
data: {"jsonrpc":"2.0","id":1,"result":{"protocolVersion":"2024-11-05","capabilities":{"experimental":{},"prompts":{"listChanged":false},"resources":{"subscribe":false,"listChanged":false},"tools":{"listChanged":false}},"serverInfo":{"name":"know-data-mcp","version":"1.26.0"}}}
MCP客户端(窗口2): 基于 sessionId 向MCP Server发起【tool调用】请求
$ curl -X POST "http://127.0.0.1:18001/messages/?session_id=$SESSION_ID" \
-H "Content-Type: application/json" \
-d '{
"jsonrpc": "2.0",
"method": "tools/call",
"params": {
"name": "minus",
"arguments": {"a": 3, "b": 4}
},
"id": 1
}'
Accepted
此时，MCP客户端(窗口1)将输出:
: ping - 2026-03-25 16:57:09.977150+00:00
event: message
data: {"jsonrpc":"2.0","id":1,"result":{"content":[{"type":"text","text":"-1"}],"structuredContent":{"result":-1},"isError":false}}
上述过程，MCP Server 将输出如下:
INFO: 127.0.0.1:62624 - "GET /sse HTTP/1.1" 200 OK
INFO: 127.0.0.1:62717 - "POST /messages/?session_id=2dcb98b6eaa74e2eade0e07160af6610 HTTP/1.1" 202 Accepted
INFO: 127.0.0.1:62743 - "POST /messages/?session_id=2dcb98b6eaa74e2eade0e07160af6610 HTTP/1.1" 202 Accepted
MCP Client端(python版)
MCP客户端的调用过程也可用 python 实现:
# mcp_sse_client.py
import asyncio
from mcp.client.sse import sse_client
from mcp import ClientSession
async def main():
async with sse_client('http://127.0.0.1:18001/sse') as streams:
async with ClientSession(*streams) as session:
await session.initialize()
tools = await session.list_tools() # 获取所有工具的信息
print(tools)
res = await session.call_tool('minus', { "a": 3, "b": 4}) # 直接调用指定工具
print(res)
if __name__ == '__main__':
asyncio.run(main())
out:
meta=None nextCursor=None tools=[Tool(name='add', title=None, description='加法运算', inputSchema={'properties': {'a': {'title': 'A', 'type': 'integer'}, 'b': {'title': 'B', 'type': 'integer'}}, 'required': ['a', 'b'], 'title': 'addArguments', 'type': 'object'}, outputSchema={'properties': {'result': {'title': 'Result', 'type': 'integer'}}, 'required': ['result'], 'title': 'addOutput', 'type': 'object'}, icons=None, annotations=None, meta=None, execution=None), Tool(name='minus', title=None, description='Subtract two numbers', inputSchema={'properties': {'a': {'title': 'A', 'type': 'integer'}, 'b': {'title': 'B', 'type': 'integer'}}, 'required': ['a', 'b'], 'title': 'minusArguments', 'type': 'object'}, outputSchema={'properties': {'result': {'title': 'Result', 'type': 'integer'}}, 'required': ['result'], 'title': 'minusOutput', 'type': 'object'}, icons=None, annotations=None, meta=None, execution=None), Tool(name='know_data', title=None, description='\n 获取know-data的信息\n\n Returns:\n know-data的信息\n ', inputSchema={'properties': {}, 'title': 'know_dataArguments', 'type': 'object'}, outputSchema={'properties': {'result': {'title': 'Result', 'type': 'string'}}, 'required': ['result'], 'title': 'know_dataOutput', 'type': 'object'}, icons=None, annotations=None, meta=None, execution=None)]
meta=None content=[TextContent(type='text', text='-1', annotations=None, meta=None)] structuredContent={'result': -1} isError=False
MCP Client端(VS Code:Copilot版)
添加 / 配置 MCP Server
默认配置文件路径: C:>User>{User}>AppData>Roaming>Code>User>mcp.json
调用 mcp tool
推荐文献
mcp server系列 part 1| vscode添加 sse协议的mcp server 保姆级教程 - Bilibili
2.1.2 Python MCP 运行流程(Server版) //todo
from mcp.server import Server
from mcp.types import TextContent
# 创建 MCP Server 实例
server = Server("my-data-server")
@server.list_resources()
async def list_resources():
"""声明可用的资源"""
return [
{"uri": "file:///data/users.json", "name": "用户数据"}
]
@server.read_resource()
async def read_resource(uri: str):
"""实现资源读取逻辑"""
if uri == "file:///data/users.json":
return TextContent(text=load_users_json())
Q:from mcp.server import Server与from mcp.server import FastMCP的区别?
参见文末 FAQ
2.1.3 通信机制
传输方式
适用场景
特点
stdio
本地子进程
安全隔离，最常见
SSE
远程服务
HTTP 流式传输
WebSocket
实时双向
低延迟交互
stdio 模式工作流程：
AI 宿主进程 ──fork──→ Python MCP Server (子进程)
↑ ↓
└─────── stdin/stdout ←─── JSON-RPC 消息
2.1.4 核心协议方法
MCP 定义了三类标准能力：
类型
方法示例
用途
Resources
resources/list, resources/read
暴露结构化数据
Tools
tools/list, tools/call
执行函数/操作
Prompts
prompts/list, prompts/get
提供模板化提示
2.1.5 实际运行示例
当 AI 需要查询数据库时：
# server.py - MCP Server 实现
from mcp.server import Server
import sqlite3
server = Server("sqlite-server")
@server.list_tools()
async def list_tools():
return [{
"name": "query_sql",
"description": "执行 SQL 查询",
"inputSchema": {
"type": "object",
"properties": {
"sql": {"type": "string"}
}
}
}]
@server.call_tool()
async def call_tool(name: str, arguments: dict):
if name == "query_sql":
conn = sqlite3.connect("data.db")
result = conn.execute(arguments["sql"]).fetchall()
return [TextContent(text=str(result))]
运行时交互序列：
1. AI 宿主启动: python server.py (子进程)
2. AI 发送: {"jsonrpc":"2.0","method":"tools/list","id":1}
3. Server 返回: 可用工具列表 (含 query_sql)
4. AI 决策: 需要调用 query_sql 查询数据
5. AI 发送: {"jsonrpc":"2.0","method":"tools/call","params":{"name":"query_sql",...},"id":2}
6. Server 执行 SQL → 返回结果 → AI 生成回答
2.1.6 Python SDK 关键组件
模块
功能
mcp.server
创建服务端
mcp.client
创建客户端连接
mcp.types
标准数据类型 (TextContent, ImageContent 等)
mcp.shared
传输层实现 (stdio, SSE)
2.1.7 设计优势
安全隔离：MCP Server 在独立进程中运行，与 AI 宿主隔离
标准统一：任何支持 MCP 的 AI 应用都能使用相同的工具
语言无关：虽然用 Python 编写，但协议是跨语言的
动态发现：AI 可以自动发现可用资源和工具，无需硬编码
2.2 MCP Server(SSE模式) - NACOS MCP Server (Python)版
使用 Nacos MCP Wrapper Python 开发 MCP Server。
step0 前置要求
安装 python / pip 环境
推荐版本: python 3.10
安装 python 依赖
$ pip install python-dotenv
$ pip install nacos-mcp-wrapper-python nacos-sdk-python
$ pip list | findstr nacos
nacos-mcp-wrapper-python 1.1.0
nacos-sdk-python 3.0.4
nacos_mcp_wrapper 是一个 MCP (Model Context Protocol) 的 Nacos 集成库。
step1 开发并启动 Mcp Server/Tool (SSE版)
.env
配置环境变量 for register_nacos_mcp_server.py
NACOS_SERVER_ADDR="127.0.0.1:8848"
NACOS_USERNAME="nacos"
NACOS_PASSWORD="nacos"
MCP_APP_SERVER_NAME="nacos-mcp-python"
MCP_APP_SERVER_PORT=18001
register_nacos_mcp_server.py
import os
from dotenv import load_dotenv
from nacos_mcp_wrapper.server.nacos_mcp import NacosMCP
from nacos_mcp_wrapper.server.nacos_settings import NacosSettings
import logging
# @description 启动 Mcp Server，并注册到 NACOS
# @dependency : pip install python-dotenv nacos-mcp-wrapper-python nacos-sdk-python
# step0 前置准备
# step0.1 加载环境变量 from `.env` 文件
load_dotenv()
# step0.2 配置日志
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s - %(levelname)s - %(message)s'
)
logger = logging.getLogger(__name__)
# step1 Create an MCP server instance
nacos_settings = NacosSettings()
# 验证所需的端口是否畅通: ssh 127.0.0.1 -p 8848 -v , ssh 127.0.0.1 -p 9848 -v
nacos_settings.SERVER_ADDR = os.getenv("NACOS_SERVER_ADDR") # <nacos_server_addr> e.g. "127.0.0.1:8848"
nacos_settings.USERNAME = os.getenv("NACOS_USERNAME") # e.g. "nacos" or ""
nacos_settings.PASSWORD = os.getenv("NACOS_PASSWORD") # e.g. "nacos" or ""
mcp = NacosMCP(
os.getenv("MCP_APP_SERVER_NAME")
, nacos_settings=nacos_settings
, version="1.0.1"
, port=os.getenv("MCP_APP_SERVER_PORT")
)
# step2 Register some tools for the mcp server / 注册工具（装饰器方式，与官方 MCP SDK 完全一致）
## Register an addition tool
@mcp.tool()
def add(a: int, b: int) -> int:
"""Add two integers together"""
return a + b
## Register a subtraction tool
@mcp.tool()
def minus(a: int, b: int) -> int:
"""Subtract two numbers"""
return a - b
## calculate tool
@mcp.tool()
def calculate(expression: str) -> str:
"""执行数学计算表达式"""
return str(eval(expression))
## get weather tool
@mcp.tool()
def get_weather(city: str) -> str:
"""获取指定城市的天气"""
return f"{city} 今天晴天，25°C"
# 运行逻辑：
# 程序启动后会注册几个工具：add / minus / calculate / get_weather / ...
# 使用 SSE 方式运行在端口 (如: 18001)（在第 9 行设置）
# 连接到 Nacos 服务注册中心 (地址：${NACOS_SERVER_ADDR}, 如: 127.0.0.1:8848)
# 服务名称为 nacos-mcp-python，版本 1.0.1
# 异常会被捕获并打印错误信息
if __name__ == "__main__":
logger.info(f"正在连接 Nacos 服务器：{nacos_settings.SERVER_ADDR}")
logger.info(f"用户名：{nacos_settings.USERNAME}")
try:
# transport/通信模式
mcp.run(transport="sse") # sse: Server-sent events/SSE (Server-sent Events)：服务器推送事件到客户端；适用于：需要实时数据推送的场景; 特点：单向通信（服务器 → 客户端），基于 HTTP 长连接
# mcp.run(transport="stdio") # stdio :标准输入输出：通过进程的 stdin/stdout 通信; 适用于：本地进程间通信、命令行工具集成; 特点：双向通信，适合本地部署
# mcp.run(transport="streamable-http") # streamable-http/流式 HTTP：支持流式响应的 HTTP 协议; 适用于：需要双向通信的 Web 场景; 特点：双向通信，支持更复杂的交互模式
except Exception as e:
error_msg = str(e)
# 如果是 404 错误，可能是首次启动，尝试继续
if "404" in error_msg and "not found" in error_msg:
logger.warning("⚠ 未在 Nacos 中找到现有配置，将创建新服务...")
# 可以选择重试或继续
else:
print(f"Runtime error: {e}")
logger.error(f"Runtime error: {e}", exc_info=True)
print(f"\n连接失败，请检查:")
print(f"1. Nacos 服务器 {nacos_settings.SERVER_ADDR} 是否可访问")
print(f"2. 用户名密码是否正确")
print(f"3. 网络防火墙是否开放端口")
查验
mcp server 端 : 日志
C:\Users\xxx\.conda\envs\ai-env\python.exe D:\Workspace\xxx\DemoPythonProject\nacos_mcp\registry_nacos_mcp.py
2026-03-25 00:28:29,937 - INFO - 正在连接 Nacos 服务器：127.0.0.1:8848
2026-03-25 00:28:29,938 - INFO - 用户名：nacos
2026-03-25 00:28:29,953 - INFO - create new rpc client: 9aec5ecb-8ea3-4262-98c5-ac0c3247b228
2026-03-25 00:28:29,953 - INFO - init app conn labels from client config,{}
2026-03-25 00:28:29,953 - INFO - init app conn labels from env,{}
2026-03-25 00:28:29,954 - INFO - final app conn labels: {}
2026-03-25 00:28:29,954 - INFO - rpc client init label, labels : {'source': 'sdk', 'module': 'ai'}
2026-03-25 00:28:29,954 - INFO - rpc client register connection listener: AIGrpcRedoService
2026-03-25 00:28:29,955 - INFO - rpc client register server push request: ConnectResetRequest handler: ConnectResetRequestHandler
2026-03-25 00:28:29,955 - INFO - rpc client register server push request: ClientDetectionRequest handler: ClientDetectionRequestHandler
2026-03-25 00:28:29,956 - INFO - rpc client start to connect server, server: 127.0.0.1:9848
2026-03-25 00:28:30,091 - INFO - [get_access_token] AccessToken: eyJhbGciOiJIUzI1NiJ9.eyJzdWIiOiJuYWNvcyIsImV4cCI6MTc3NDM4NzcxMH0.oLQo0kzFuNDOTQlUpQXcm4QqEE-gNIYjWepdGlXs2-4, TTL: 18000, force_refresh: True
2026-03-25 00:28:30,099 - INFO - connect to server success,labels:{'source': 'sdk', 'module': 'ai'},tenant:public,connection_id:1774369710005_183.xxx.202.190_7436
2026-03-25 00:28:30,114 - INFO - receive stream server request, connection_id:1774369710005_183.xxx.202.190_7436, original info: metadata {
type: "SetupAckRequest"
clientIp: "172.17.0.2"
}
body {
value: "{\"headers\":{},\"abilityTable\":{\"supportPersistentInstanceByGrpc\":true,\"fuzzyWatch\":true,\"lock\":true,\"mcp\":true},\"module\":\"internal\"}"
}
2026-03-25 00:28:30,218 - INFO - rpc client successfully connected to server:127.0.0.1:9848, connection_id:1774369710005_183.xxx.202.190_7436
2026-03-25 00:28:30,223 - INFO - rpc client notify [connected] event to listeners
2026-03-25 00:28:30,530 - ERROR - failed to invoke nacos config server : Error [404]: MCP server `nacos-mcp-python` not found in namespaceId: `public`
2026-03-25 00:28:30,531 - INFO - can not found McpServer info from nacos,nacos-mcp-python,version:1.0.1
2026-03-25 00:28:30,552 - ERROR - failed to invoke nacos config server : Error [404]: MCP server `nacos-mcp-python` not found in namespaceId: `public`
2026-03-25 00:28:30,553 - INFO - [9aec5ecb-8ea3-4262-98c5-ac0c3247b228] release mcp server: nacos-mcp-python, version 1.0.1
2026-03-25 00:28:30,890 - INFO - mcp_id:ea9e94c3-18f2-48a7-ae4e-9de982ea8718, namespace_id:public, mcp_name:nacos-mcp-python changed, mcp_server_detail_info:<class 'v2.nacos.ai.model.mcp.mcp.McpServerDetailInfo'>
2026-03-25 00:28:30,890 - INFO - Register to nacos success,nacos-mcp-python,version:1.0.1
INFO: Started server process [28956]
INFO: Waiting for application startup.
INFO: Application startup complete.
INFO: Uvicorn running on http://127.0.0.1:18001 (Press CTRL+C to quit)
2026-03-25 00:28:41,025 - INFO - mcp service changed: nacos-mcp-python -> {"id":"ea9e94c3-18f2-48a7-ae4e-9de982ea8718","name":"nacos-mcp-python","protocol":"mcp-sse","frontProtocol":"mcp-sse","description":"nacos-mcp-python","repository":null,"packages":null,"versionDetail":{"version":"1.0.1","release_data":null,"is_latest":true},"version":"1.0.1","remoteServerConfig":{"serviceRef":{"namespaceId":"public","groupName":"DEFAULT_GROUP","serviceName":"nacos-mcp-python::1.0.1","transportProtocol":null},"exportPath":"/sse","frontEndpointConfigList":[]},"localServerConfig":null,"enabled":true,"status":"active","capabilities":["TOOL"],"backendEndpoints":[{"protocol":null,"address":"192.168.60.18","port":18001,"path":"/sse","headers":null}],"frontendEndpoints":[],"toolSpec":{"specificationType":null,"encryptData":null,"tools":[{"name":"add","description":"Add two integers together","inputSchema":{"properties":{"a":{"title":"A","type":"integer"},"b":{"title":"B","type":"integer"}},"required":["a","b"],"title":"addArguments","type":"object"}},{"name":"minus","description":"Subtract two numbers","inputSchema":{"properties":{"a":{"title":"A","type":"integer"},"b":{"title":"B","type":"integer"}},"required":["a","b"],"title":"minusArguments","type":"object"}},{"name":"calculate","description":"执行数学计算表达式","inputSchema":{"properties":{"expression":{"title":"Expression","type":"string"}},"required":["expression"],"title":"calculateArguments","type":"object"}},{"name":"get_weather","description":"获取指定城市的天气","inputSchema":{"properties":{"city":{"title":"City","type":"string"}},"required":["city"],"title":"get_weatherArguments","type":"object"}}],"toolsMeta":{},"securitySchema":null},"allVersions":[{"version":"1.0.1","release_data":null,"is_latest":true}],"namespaceId":"public"}
2026-03-25 00:28:41,025 - INFO - mcp_id:ea9e94c3-18f2-48a7-ae4e-9de982ea8718, namespace_id:public, mcp_name:nacos-mcp-python changed, mcp_server_detail_info:<class 'v2.nacos.ai.model.mcp.mcp.McpServerDetailInfo'>
nacos server 端 : MCP管理 - MCP 列表 (可选步骤)
MCP Server 并不依赖 NACOS Server，只是说 NACOS Server(3.x起) 为 MCP Server 提供了 MCP 服务注册与发现的能力。
step2 客户端(curl为例): 请求 mcp server 以获得 SessionId
$ curl -N http://127.0.0.1:18001/sse
event: endpoint
data: /messages/?session_id=1deb0a8254604faf8416fc7beb540604
: ping - 2026-03-24 15:29:28.860631+00:00
...
注意：-N 参数至关重要，它告诉 curl 禁用缓冲，立即显示服务器推过来的数据。
step2 客户端(curl为例): 新建第2个窗口/进程，基于 SessionId 请求 mcp server 完成【会话初始化】操作
为什么 MCP 客户端需要2个窗口？
MCP (Model Context Protocol) 在使用 SSE (Server-Sent Events) 传输时，采用了 “读写分离” 的模式：
读 (Read): 一个持久的 HTTP 长连接，服务器通过它源源不断地推送数据（Events）。
写 (Write): 短暂的 HTTP POST 请求，客户端通过它发送指令。
因此，我们需要打开两个终端窗口：一个负责“听”，一个负责“说”。
$ SESSION_ID=1deb0a8254604faf8416fc7beb540604
echo "${SESSION_ID}"
1deb0a8254604faf8416fc7beb540604
$ curl -X POST "http://127.0.0.1:18001/messages/?session_id=$SESSION_ID" \
-H "Content-Type: application/json" \
-d '{
"jsonrpc": "2.0",
"id": 1,
"method": "initialize",
"params": {
"protocolVersion": "2024-11-05",
"capabilities": {},
"clientInfo": { "name": "test-client", "version": "1.0.0" }
}
}'
注：Body: 标准的 JSON-RPC 2.0 格式。
MCP 客户端 (2个窗口):
MCP 服务器端
类似日志: "INFO: 127.0.0.1:65285 - "GET /sse HTTP/1.1" 200 OK"
注：如果不完成此初始化操作，直接调用 tool ，将调用失败，报类似错误:
"WARNING - Failed to validate request: Received request before initialization was complete"
"data: {"jsonrpc":"2.0","id":1,"error":{"code":-32602,"message":"Invalid request parameters","data":""}}"
step3 客户端(curl为例): 在第2个窗口/进程，基于 SessionId 调用 mcp server 的指定 tool
$ curl -X POST "http://127.0.0.1:18001/messages/?session_id=$SESSION_ID" \
-H "Content-Type: application/json" \
-d '{
"jsonrpc": "2.0",
"method": "tools/call",
"params": {
"name": "minus",
"arguments": {"a": 3, "b": 4}
},
"id": 1
}'
注：Body: 标准的 JSON-RPC 2.0 格式。
调用其他 tool 的示范
Z FAQ/Troubleshooting for MCP
Q: 发送 POST 后没反应？
检查 客户端窗口1： 结果是异步返回的，一定要看监听窗口。
检查斜杠：确认 POST URL 是 /messages/?... 而不是 /messages?...。可以加 -v 参数查看是否返回了 307 Temporary Redirect。
Q: Session ID 过期？
SSE 连接如果断开，Session ID 就会失效。每次重新运行 curl -N .../sse 都会生成一个新的 ID，发 POST 时记得更新。
Q: 乱码或无输出？
确保 curl 加上了 -N (no buffer)。
确保 curl 加上了 -H "Content-Type: application/json"，否则服务器可能不解析 Body。
Q: Python MCP 中 from mcp.server import Server与from mcp.server import FastMCP的区别?`
这两个都是 MCP SDK 中用于创建 MCP 服务器的类，但定位和使用方式不同。
Server - 基础服务器类
特点：
底层、原始的服务器实现
需要手动处理更多细节
更灵活，可以完全自定义行为
类似 Flask 的原始 API
使用示例：
from mcp.server import Server
from mcp.types import Tool
# 创建服务器实例
server = Server("my-mcp-server")
# 手动注册工具
@server.call_tool()
async def add(arguments: dict) -> list:
a = arguments.get("a", 0)
b = arguments.get("b", 0)
return [{"type": "text", "text": str(a + b)}]
"""或:如下实现 add tool
@server.call_tool()
async def call_add(name: str, arguments: dict):
if name != "add":
raise ValueError(f"Unknown tool: {name}")
result = arguments["a"] + arguments["b"]
return [{ "type": "text", "text": str(result) }]
"""
# 手动定义工具列表
@server.list_tools()
async def list_tools() -> list[Tool]:
return [
Tool(
name="add",
description="Add two numbers",
inputSchema={
"type": "object",
"properties": {
"a": {"type": "integer"},
"b": {"type": "integer"}
},
"required": ["a", "b"]
}
)
]
if __name__ == "__main__":
server.run()
FastMCP - 快速开发服务器
特点：
高级封装，类似 FastAPI 的风格
自动推断工具 schema（从函数签名）
装饰器语法更简洁
自动处理类型转换和验证
推荐使用 ⭐
from mcp.server import FastMCP
# 创建服务器（自动命名）
app = FastMCP('know-data-mcp', port=12345)
@app.tool()
def add(a: int, b: int) -> int:
"""Add two integers"""
return a + b
# 使用装饰器，自动注册工具并生成 schema
@app.tool()
async def know_data() -> str:
"""获取 know-data 的信息"""
return "know-data 是个喜欢捣鼓技术的软件工程师"
# 不需要手动定义 list_tools，框架会自动处理
if __name__ == "__main__":
app.run(transport='sse')
核心区别对比表
特性
Server
FastMCP
抽象级别
底层
高级封装
代码量
较多
简洁
Schema 生成
手动定义
自动生成
类型检查
手动处理
自动推断
学习曲线
陡峭
平缓
灵活性
高
中等
推荐场景
复杂定制需求
快速开发 ⭐
总结
Python 中 ✅ 优先使用 FastMCP，因为：
开发效率高
代码更可读
自动类型安全
符合 Pythonic 风格
只有在以下情况才使用 Server：
需要完全自定义协议行为
实现非标准的 MCP 扩展
对性能有极致要求
Q: MCP 的 SSE 模式，是否用到了 Websocket 技术?
SSE（Server-Sent Events）和 WebSocket 是两种不同的技术，SSE 模式并不使用 WebSocket。
它们的区别：
SSE vs WebSocket 对比
特性
SSE
WebSocket
通信方向
单向：服务器 → 客户端
双向：客户端 ↔ 服务器
协议基础
HTTP（长连接）
WebSocket 协议（ws://）
连接建立
标准 HTTP GET 请求
HTTP Upgrade 握手后切换
数据格式
纯文本，以 data: 开头的流
二进制或文本帧
自动重连
浏览器原生支持
需手动实现
MCP 用途
Server → Client 推送
双向实时通信
MCP 中的 SSE 模式原理
┌─────────────┐ ┌─────────────┐
│ MCP Client │ ──HTTP GET /sse──→│ MCP Server │
│ (AI 宿主) │ ←──text/event-stream│ (Python) │
└─────────────┘ └─────────────┘
↑ ↓
└────────HTTP POST /message────────┘
(客户端发送请求)
SSE 数据流格式：
HTTP/1.1 200 OK
Content-Type: text/event-stream
Cache-Control: no-cache
Connection: keep-alive
data: {"jsonrpc":"2.0","method":"resources/list","id":1}
data: {"jsonrpc":"2.0","result":[...],"id":1}
data: {"jsonrpc":"2.0","method":"notifications/resources/updated",...}
MCP 三种传输方式的技术栈
MCP 传输方式
底层技术
双向通信实现
stdio
操作系统管道 (stdin/stdout)
两个单向流组合
SSE
HTTP + SSE 流
SSE 流 + 独立 HTTP POST 请求
WebSocket
WebSocket 协议
原生双向帧
Python MCP SSE 服务端示例
from mcp.server.sse import SseServerTransport
from starlette.applications import Starlette
from starlette.routing import Route
# SSE 传输层
sse = SseServerTransport("/messages")
async def handle_sse(request):
"""处理 SSE 连接：服务器推送消息给客户端"""
async with sse.connect_session(request) as streams:
await server.run(streams[0], streams[1], server.create_initialization_options())
async def handle_message(request):
"""处理客户端 POST 请求"""
await sse.handle_post_message(request)
return Response(status_code=202)
app = Starlette(routes=[
Route("/sse", handle_sse), # 建立 SSE 连接
Route("/messages", handle_message, methods=["POST"]) # 接收客户端消息
])
关键区别总结
SSE 模式的双向通信是这样实现的：
Server → Client：SSE 流（text/event-stream）
Client → Server：独立的 HTTP POST 请求
WebSocket 模式：
单个 ws:// 连接同时处理双向通信
MCP 协议设计时，SSE 和 WebSocket 是并列的两种传输选项，不是包含关系。Python mcp SDK 中：
mcp.server.sse → SSE 实现
mcp.server.websocket → WebSocket 实现（如果存在）
如果你需要真正的双向实时流（比如高频数据推送），WebSocket 更合适；如果只是 AI 工具调用场景，SSE 足够且更简单。
Q:如何在 VSCode中查看 指定 MCP Server中有哪些 Tool?
方法1：使用 MCP Inspector（推荐）
MCP Inspector 是官方提供的调试工具，可以图形化查看所有 Tools：
# 安装 inspector
npx @modelcontextprotocol/inspector <command>
在 VSCode 终端中运行后，会自动打开浏览器界面，清晰展示：
Tools 列表及参数
Resources 资源
Prompts 提示模板
方法2：VSCode 扩展（未来支持）
目前 VSCode 官方正在开发 MCP 原生支持，预计会提供：
内置 MCP Server 管理面板
Tools 自动发现与补全
直接调用 Tools 的界面
方法3: 接入/集成 NACOS / Higress 网关等集成产品中。
Y 推荐文献
[AIGC/Agent] MCP := 模型上下文协议 := AI Agent应用与外部系统集成的标准协议 - 博客园/千千寰宇 【推荐】
[Nacos/Docker/MCP] Nacos 3.x : 为 AI MCP 而生 - 博客园/千千寰宇 【推荐】
SSE(Server Send Events) ：服务器 => 浏览器的消息推送解决方案 - 博客园/千千寰宇
[JSON/RPC/MCP] JSON-RPC 2.0 : 轻量级远程过程调用协议 - 博客园/千千寰宇
[Web/Python] ASGI : 异步 Python Web 网关服务器 - 博客园/千千寰宇
NACOS
MCP Server 自动注册与发现手册 - NACOS
Nacos MCP Router手册 - NACOS
MCP模版配置手册 - NACOS
第三方
https://open-mcp-client.vercel.app
一个免费的mcp在线客户端
mcp server系列 part 1| vscode添加 sse协议的mcp server 保姆级教程 - Bilibili
X 参考文献
如何使用 curl 命令行测试 FastMCP (SSE 模式) - CSDN 2026.01.03
Model Context Protocol(MCP)的SSE使用 - blog.kala.love //todo
pip install mcp[cli]
from mcp.server import FastMCP
如何搭建sse的MCP本地服务器以及Cursor应用 - ZhihuSSE/Server-sent events/SSE (Server-sent Events)： 服务器推送事件到客户端；