mcp client-server开发到部署
原创
mcp client-server开发到部署
原创
—— 01 为什么需要MCP
谈到mcp的必要性，我们不得不先说function call。function call其实就是大模型的工具库，能够扩展大模型的知识边界。但之前我也存在一些疑问：RAG不能做到吗？联网搜索不能做到吗？在知识扩展的能力上，他们三者对于大模型都能有增益，但各自擅长的点不一样。
RAG一定程度上能解决大模型的幻读，提高大模型的输出质量，但因为RAG技术方案的局限性（基于内容的相似检索），检索到的内容比较片面。一句话概括就是缺少大局观，比如从商品知识库中检索出最低价格这类问题，需要通读全文才能得到最佳答案，RAG的表现就不是很好。
而联网搜索由于搜索内容非结构化，同时需要对检索内容作进一步清洗（不一定检索到的都是高质量内容），比如社区论坛、广告内容等对于大模型而言都存在一定对信息噪音，降低模型输出内容的准确率。
而function call工具能力对于知识获取的准确率、以及解决一些特定领域的问题，就存在较大的优势。但因为缺少统一标准，开发者往往需要针对每一个模型或者工具都单独开发一套代码，使得function call的开发和维护成本比较高。因此就有了近期比较火热的MCP概念，MCP全称叫Model Context Protocol（模型上下文协议），在Function call的基础上提供了结构化的标准，统一工具输入输出。以下是从官方文档中截取的图，很好的说明了MCP的位置。
那么MCP和大模型到底是怎么通信的呢？通过以下时序图，我们能更清晰的感知到用户- Agent - LLM - MCP Server的交互流程。
—— 02 快速体验MCP
腾讯云开发者社区MCP广场目前已经提供了较多的MCP SERVER，能了解到大量Server本地部署的方式，同时还支持云托管模型（当前免费），能快速部署远程MCP Server。
同时，结合CodeBuddy（充当MCP Host With MCP Client）提供的MCP安装脚本，可以快速体验下MCP Server在生产中实际带来的效果。
mcp 的通信方式总共有三种：stdio、sse、streamableHttp，stdio是本地部署的方式，后面两种支持远程访问。结合前面时序图的理解，以及本节的快速体验，我们对于mcp交互方式应该都有一个大概的认知，但是mcp client- server之间的通信细节目前我们是不清楚的，大多数场景包括ide、代码助手等，为了用户方便已经做好了client的封装。
—— 03 Client&Server 开发(stdio)
那么对于mcp client和server是怎么实现的呢？下面我们还是以腾讯地图提供的一个工具查询（ip定位）为例，展示下client&server之间交互的逻辑细节。
正式展示之前我们先用uv创建好项目：
## 安装uv工具，uv 是一款包管理工具，速度快&环境隔离
pip install uv
## 用uv初始化一个项目
uv init get_location
cd get_location
## 进入项目所在目录，创建虚拟环境
uv .venv
## 安装必要的依赖（根据开发需求可以随时增加）
uv add mcp httpx openai3.1 server
server的代码逻辑比较简单，核心是提供给client一个tool说明和API调用
from mcp.server.fastmcp import FastMCP
3.2 client
客户端初始化及连接建立
发起工具调用流程
运行方式及结果
—— 04 SSE/HTTP方式的改写及DEBUG
以上展示stdio传输方式的client-server，如果想改成sse的方式应该怎么做？
4.1 server
针对server端改动比较小，直接修改transport就可以
对于client端的改动也不是很大，需要修改client的连接方式
启动方式上也有一些区别，server需要优先启动
其次client启动时配置server的访问地址
运行client并进行简单测试
4.2 streamable http(流式http)
最新SDK推出的第三种通信协议，相比sse的方式，流式http的方式则支持双向通信，整体性能更加可靠，同时代码改动量也不是很大。
相较于前两种方式，client端代码同样也是需要修改connect方法，核心逻辑如下：
同样，server端改动也很小，仅需要修改run参数
服务启动方式，直接使用Python命令即可
对应客户端启动及简单的问答测试
4.3 如何进行debug
mcp官方提供了调试工具：inspector，调试工具的安装依赖
node version ≥v22.7.5
npx version ≥9.x
启动方式：npx @modelcontextprotocol/inspector
我们以streamable HTTP方式启动mcp server为例，演示下工具如何使用
工具调试
—— 05 Pypi平台部署
当你完成server服务开发后，为了让其他开发者更便捷的使用到你的工具，我们可以考虑将server打包发布到pypi平台。
打包发布需要在pypi平台上注册账号，注册流程如下：
● 进入Account settings页面
● 拉到最下面，进行身份二次验证（创建token的前提），其中TOTPAuthenticator APP需要提前下载（iOS可以在应用市场搜索，安卓用户可以到谷歌应用市场搜索，部分国内应用市场可能没有收录）
● 回到Account settings页面，找到Add token并开始添加
● 点击Create token之后，将token复制到指定目录下
● 到这就完成了token配置，后续上传包就可以自动校验，接下来执行打包流程
******创建项目及代码结构
uv init mcp_server ## 初始化项目
cd mcp_server ## 进去项目主目录
uv venv ## 创建虚拟环境
source .venv/bin/activate ## 激活虚拟环境
uv add mcp httpx ## 安装依赖库
mkdir -p src/mcp_server ## 打包目录结构 核心代码放到这个路径下
******创建server
cd src/mcp_server
touch server.py ## mcp 服务主逻辑代码
touch __init__.py ## 初始化 文件包含一行代码：from .server import main
******回到项目主目录开始打包
cd ../../
uv build
******构建成功后当前路径下会产生dist的文件夹，代码构建成功，执行上传
******--repository 参数代表指定仓库 当前我们是上传到测试平台
twine upload --repository testpypi dist/ #发布测试平台
twine upload dist/ #发布正式平台● 上传包成功之后，就能拿到对应的下载地址
● 验证包是否可用
—— 06 Client端支持多Server完整代码展示
通过以上client-Server的拆解，其实要实现一个“hello world”版MCP Server是很简单的事情，更多需要大家根据具体的业务逻辑去实现，而client端相对会复杂些，需要适配不同的MCP-Server通信方式。好在官方提供的SDK也是十分便捷的，以下将模拟一个MCP Host的实现，支持MCP Server任意通信方式(stdio/sse/streamable http)的调用（运行时需修改ClientConfig相关参数）。
import json
import asyncio
from dataclasses import dataclass
from typing import Optional, List, Dict, Any, TypedDict
from contextlib import AsyncExitStack
import httpx
from openai import AsyncOpenAI
from openai.types.chat import ChatCompletionMessageToolCall
from mcp import ClientSession, StdioServerParameters
## sse 特有
from mcp.client.sse import sse_client
## stdio 特有
from mcp.client.stdio import stdio_client
## streamablehttp 特有
from mcp.client.streamable_http import streamablehttp_client
from datetime import timedelta
# ----------------------------
# 配置管理
# ----------------------------
# ============= 配置结构定义 =============
class ServerConfig(TypedDict):
type: str # 'stdio', 'sse', 或 'streamable_http'
url: Optional[str] # SSE 或 streamable HTTP 的URL
command: Optional[str] # stdio的命令
args: Optional[List[str]] # stdio的参数
env: Optional[Dict[str, str]] # stdio的环境变量
class ClientConfig(TypedDict):
openai_api_key: str
base_url: str
model: str
servers: Dict[str, ServerConfig] # 服务器名称到配置的映射
# ============= 服务配置 =============
CONFIG: ClientConfig = {
## 大模型api key
"openai_api_key": "请修改成自己的server",
## 大模型访问地址 需要修改成对应厂商的地址
"base_url": "https://api.hunyuan.cloud.tencent.com/v1",
## 大模型名称 需要修改成你实际调用的模型名称
"model": "hunyuan-lite",
## 多个服务器配置 根据实际需要增删 以下仅是demo
# 可通过腾讯云社区MCP广场部署多个服务器 https://cloud.tencent.com/developer/mcp
"servers": {
"fetch_server_sse": {
"type": "sse",
## 腾讯云社区MCP广场托管的MCP服务
"url": "https://mcp-api.tencent-cloud.com/sse/*****",
"command": None,
"args": None,
"env": None
},
"location_server_http": {
"type": "streamable_http",
## 本地部署的HTTP服务
"url": "http://localhost:8000/mcp",
"command": None,
"args": None,
"env": None
},
"local_stdio": {
"type": "stdio",
"command": "python3",
"args": ["server.py"],
"env": None,
"url": None
}
# 可以添加更多服务器配置
}
}
# ----------------------------
# 核心客户端类
# ----------------------------
class MCPClient:
def __init__(self):
self.ai_client = AsyncOpenAI(
api_key=CONFIG["openai_api_key"],
base_url=CONFIG["base_url"]
)
# 存储每个服务器的会话
self.sessions: Dict[str, ClientSession] = {}
self.exit_stack = AsyncExitStack()
# 存储每个服务器的工具信息
self.server_tools: Dict[str, List[Any]] = {}
async def connect_server(self, server_name: str, server_config: ServerConfig):
"""连接到指定的MCP服务器"""
try:
if server_config["type"] == "stdio":
server_params = StdioServerParameters(
command=server_config["command"],
args=server_config["args"],
env=server_config["env"]
)
stdio_transport = await self.exit_stack.enter_async_context(stdio_client(server_params))
stdio, write = stdio_transport
session = await self.exit_stack.enter_async_context(ClientSession(stdio, write))
elif server_config["type"] == "sse":
sse_context = sse_client(url=server_config["url"])
streams = await self.exit_stack.enter_async_context(sse_context)
session = await self.exit_stack.enter_async_context(ClientSession(*streams))
elif server_config["type"] == "streamable_http":
stream_context = streamablehttp_client(
url=server_config["url"],
timeout=timedelta(seconds=60),
)
stream_transport = await self.exit_stack.enter_async_context(stream_context)
read_stream, write_stream, get_session_id = stream_transport
session = await self.exit_stack.enter_async_context(ClientSession(read_stream, write_stream))
# 如果有session_id，打印出来
session_id = get_session_id()
if session_id:
print(f"Session ID: {session_id}")
else:
raise ValueError(f"不支持的服务器类型: {server_config['type']}")
await session.initialize()
self.sessions[server_name] = session
print(f"\n成功连接到 MCP 服务器 {server_name} ({server_config['type']})")
if server_config["url"]:
print(f"URL: {server_config['url']}")
# 获取并存储工具列表
tools = (await session.list_tools()).tools
self.server_tools[server_name] = tools
# 打印工具列表（链接测试）
print(f"\n{server_name} 可用工具列表:")
for tool in tools:
print(f"- {tool}")
return True
except Exception as e:
print(f"连接 {server_name} 失败: {str(e)}")
return False
async def connect_all_servers(self):
"""连接到所有配置的服务器"""
connected_count = 0
for server_name, server_config in CONFIG["servers"].items():
if await self.connect_server(server_name, server_config):
connected_count += 1
if connected_count == 0:
raise ConnectionError("没有成功连接到任何MCP服务器")
print(f"\n成功连接到 {connected_count}/{len(CONFIG['servers'])} 个MCP服务器")
async def _get_tools_schema(self) -> List[Dict]:
"""生成所有服务器的工具模式"""
all_tools = []
for server_name, session in self.sessions.items():
tools = self.server_tools[server_name]
for tool in tools:
all_tools.append({
"type": "function",
"function": {
"name": f"{server_name}.{tool.name}", # 添加服务器名称前缀
"description": f"[{server_name}] {tool.description}", # 在描述中添加服务器标识
"parameters": tool.inputSchema
}
})
return all_tools
async def _execute_tool_call(self, tool_call: ChatCompletionMessageToolCall) -> str:
"""执行工具调用并返回结果"""
try:
# 解析服务器名称和工具名称
full_name = tool_call.function.name
if "." not in full_name:
return f"错误: 工具名称格式不正确，应为 'server_name.tool_name'"
server_name, tool_name = full_name.split(".", 1)
if server_name not in self.sessions:
return f"错误: 未找到服务器 {server_name}"
args = json.loads(tool_call.function.arguments)
print(f"\n调用工具 {server_name}.{tool_name} 参数: {args}")
# 调用对应服务器的工具
result = await self.sessions[server_name].call_tool(
tool_name,
args
)
return result.content[0].text if result.content else ""
except Exception as e:
return f"工具执行失败: {str(e)}"
async def generate_response(self, messages: List[Dict]) -> str:
"""生成 AI 响应（含工具调用处理）"""
# 首次请求（可能触发工具调用）
response = await self.ai_client.chat.completions.create(
model=CONFIG["model"],
messages=messages,
tools=await self._get_tools_schema(),
tool_choice="auto"
)
print(f"\n第一次请求messages: {messages}")
print(f"模型返回：{response}")
msg = response.choices[0].message
# 处理工具调用
if msg.tool_calls:
print("检测到工具调用请求")
messages.append(msg.model_dump(include={'role', 'content', 'tool_calls'}))
# 执行所有工具调用
for tool_call in msg.tool_calls:
tool_result = await self._execute_tool_call(tool_call)
messages.append({
"role": "tool",
"content": tool_result,
"tool_call_id": tool_call.id
})
print(f"\ntool_result: {tool_result}")
print(f"\n第二次请求messages: {messages}")
# 第二次请求（生成最终响应）
final_response = await self.ai_client.chat.completions.create(
model=CONFIG["model"],
messages=messages
)
return final_response.choices[0].message.content
return msg.content or "未获取到有效响应"
async def chat_loop(self):
"""交互式聊天循环"""
print("\n输入 'exit' 退出对话")
print("输入 'servers' 查看已连接的服务器")
print("输入 'tools' 查看所有可用工具")
while True:
try:
query = input("\n[用户]: ").strip()
if query.lower() in ('exit', 'quit'):
break
elif query.lower() == 'servers':
print("\n已连接的服务器:")
for server_name in self.sessions.keys():
print(f"- {server_name}")
continue
elif query.lower() == 'tools':
print("\n所有可用工具:")
for server_name, tools in self.server_tools.items():
print(f"\n{server_name}:")
for tool in tools:
print(f"- {tool.name}: {tool.description}")
continue
response = await self.generate_response([
{"role": "user", "content": query}
])
print(f"\n[助手]: {response}")
except KeyboardInterrupt:
break
except Exception as e:
print(f"处理出错: {str(e)}")
async def cleanup(self):
"""清理资源"""
await self.exit_stack.aclose()
print("\n所有连接已关闭")
# ----------------------------
# 主程序入口
# ----------------------------
async def main():
if not CONFIG["openai_api_key"]:
raise ValueError("请设置openai_api_key")
client = MCPClient()
try:
# 连接所有配置的服务器
await client.connect_all_servers()
await client.chat_loop()
finally:
await client.cleanup()
if __name__ == "__main__":
asyncio.run(main())—— 07 小结
当前MCP的概念比较火热，本文通过一个简单的MCP Client尝试还原整个交互过程，随着MCP工具数量的急速增长，并不是所有工具都适合加入到我们的Client中。本质上，MCP对于大模型而言是一个prompt的存在，因此较多的工具对于模型而言不一定结果就是正向的。工具越多，通信间的token消耗越多，而且未经筛选的工具如果同质化严重，模型对于这类工具的选择不一定是最优的。而且模型对于工具的选择和参数的拼接，仍然会有概率出错（跟模型本身的训练有关），这种概率出错会在降低用户体验的同时，加剧token消耗（反复调用）。因此，当前对于MCP Server的灵活使用个人感觉还是比较依赖开发者自身。
原创声明：本文系作者授权腾讯云开发者社区发表，未经许可，不得转载。
如有侵权，请联系 cloudcommunity@tencent.com 删除。
原创声明：本文系作者授权腾讯云开发者社区发表，未经许可，不得转载。
如有侵权，请联系 cloudcommunity@tencent.com 删除。
腾讯云开发者
