[AI/应用/MCP] MCP Server/Tool 开发指南

杭环 · 昨天 11:10

0 序

MCP := 模型上下文协议 := AI Agent应用与外部系统集成的标准协议

一言以蔽之，MCP = AI LLM(大语言模型) 与外部系统的集成协议

本文旨在总结和分享 MCP Server 和 MCP Tool 的开发经验。

总结踩坑的日志。

旨在用最精简的方式剖析/展示 MCP 应用的开发过程细节。

欢迎交流。

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/SSE (Server-sent Events)：服务器推送事件到客户端；

适用于：需要实时数据推送的场景; 特点：单向通信（服务器 → 客户端），基于 HTTP 长连接

JSON-RPC

2 实践案例

2.1 Python 原生版 MCP（Model Context Protocol）

安装 Python MCP

pip install mcp[cli]

复制代码

Python MCP 运行流程

1. 初始化阶段

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())

复制代码

2. 通信机制

传输方式适用场景特点stdio本地子进程安全隔离，最常见SSE远程服务HTTP 流式传输WebSocket实时双向低延迟交互stdio 模式工作流程：

AI 宿主进程 ──fork──→ Python MCP Server (子进程)
↑ ↓
└─────── stdin/stdout ←─── JSON-RPC 消息

复制代码

3. 核心协议方法

MCP 定义了三类标准能力：
类型方法示例用途Resourcesresources/list, resources/read暴露结构化数据Toolstools/list, tools/call执行函数/操作Promptsprompts/list, prompts/get提供模板化提示4. 实际运行示例

当 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 生成回答

复制代码

Python SDK 关键组件

模块功能mcp.server创建服务端mcp.client创建客户端连接mcp.types标准数据类型 (TextContent, ImageContent 等)mcp.shared传输层实现 (stdio, SSE)设计优势

安全隔离：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: MCP 的 SSE 模式，是否用到了 Websocket 技术?

SSE（Server-Sent Events）和 WebSocket 是两种不同的技术，SSE 模式并不使用 WebSocket。

它们的区别：

SSE vs WebSocket 对比

特性SSEWebSocket通信方向单向：服务器 → 客户端双向：客户端 ↔ 服务器协议基础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)两个单向流组合SSEHTTP + SSE 流SSE 流 + 独立 HTTP POST 请求WebSocketWebSocket 协议原生双向帧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 足够且更简单。

Y 推荐文献

[AIGC/Agent] MCP := 模型上下文协议 := AI Agent应用与外部系统集成的标准协议 - 博客园/千千寰宇【推荐】

[Nacos/Docker/MCP] Nacos 3.x : 为 AI MCP 而生 - 博客园/千千寰宇【推荐】
NACOS

MCP Server 自动注册与发现手册 - NACOS

Nacos MCP Router手册 - NACOS

MCP模版配置手册 - NACOS

第三方

https://open-mcp-client.vercel.app

一个免费的mcp在线客户端

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应用 - Zhihu

来源：程序园用户自行投稿发布，如果侵权，请联系站长删除
免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！

账号		自动登录	找回密码
密码			立即注册

[AI/应用/MCP] MCP Server/Tool 开发指南

相关帖子

签约作者

[AI/应用/MCP] MCP Server/Tool 开发指南

相关帖子

相关推荐

签约作者