前言
上一篇文章（手把手MCP教学-基础学习和实践）我们学习MCP基础的知识，并实操了服务端和客户端的构建和测试，但这远远不是MCP的能力，MCP协议不仅仅是让LLM能够使用一种工具，而是应该让LLM能够连接多个服务器，使用多种工具。
这篇文件将帮助你构建一个能够连接多个服务器的MCP客户端。
服务端
为了实现我们的MCP服务端连接多个客户端，我们首先需要增加多一个服务端，这里我增加多一个python文件控制服务器，使得LLM通过天气插件和文件控制插件实现获取天气并以文件形式保存到我电脑中的这样一个多工具使用的能力。
文件控制服务器
1. 初始化
| 这里默认你已经有python和uv环境，如果你电脑中还没有环境，请自行配置环境，不会的话可以查看上一篇文章 |
| powershell
# 初始化新的项目名字为filesystem
cd server
uv init filesystem
cd filesystem
# 创建并进入虚拟环境
uv venv
.venvScriptsctivate
# 安装依赖项
uv add mcp[cli]
# 创建服务项文件
new-item filesystem.py（可以自己手动新建一个py文件，如果你使用的终端没有new-item指令的话） |
完成后项目目录如下：
2. 构建文件控制服务器
导入包并设置实例
将这些添加到您的顶部weather.py：
| python
from mcp.server.fastmcp import FastMCP
# 初始化FastMCP服务器
mcp = FastMCP("filesystem") |
实现工具执行
工具执行程序负责实际执行每个工具的逻辑：
| python
@mcp.tool()
async def create_file(file_name: str, content: str) -> str:
"""
创建文件
:param file_name: 文件名
:param content: 文件内容
:return: 创建成功消息
"""
try:
with open(file_name, "w", encoding="utf-8") as file:
file.write(content)
return f"文件'{file_name}'创建成功"
except Exception as e:
return f"创建文件失败: {str(e)}"
@mcp.tool()
async def read_file(file_name: str) -> str:
"""
读取文件内容
:param file_name: 文件名
:return: 文件内容或错误消息
"""
try:
with open(file_name, "r", encoding="utf-8") as file:
return file.read()
except FileNotFoundError:
return f"文件'{file_name}'未找到"
except Exception as e:
return f"读取文件失败: {str(e)}"
@mcp.tool()
async def write_file(file_name: str, content: str) -> str:
"""
写入文件内容
:param file_name: 文件名
:param content: 文件内容
:return: 写入成功消息
"""
try:
with open(file_name, "w", encoding="utf-8") as file:
file.write(content)
return f"文件'{file_name}'写入成功"
except Exception as e:
return f"写入文件失败: {str(e)}"
|
运行服务器
最后，我们初始化并运行服务端：
| PowerShell
if __name__ == "__main__":
# 以标准 I/O 方式运行 MCP 服务器
mcp.run(transport="stdio") |
您的服务器已完成！运行uv run filesystem.py确认一切正常。
3. 运行天气预报服务器
新建一个终端：
| PowerShell
cd server
cd weather
uv venv
.venvScriptsctivate
uv run weather.py |
确保运行成功。
改造客户端
客户端想要实现交互式的LLM多工具调用，主要需要增加以下能力：
- 从单个服务器引入，变成多个服务器引入，并且将多个服务器的工具整合一并提供给LLM
- 对话是否结束交由LLM决定，当调用一次工具后不足以让LLM完成本次任务，将循环让LLM调用工具，直至LLM确定不需要再调用工具，则返回。
确定了这两个关键但直接上代码：
在client的mcp-client目录下新建一个client_tools.py文件：
| python
import asyncio
import os
from openai import OpenAI
from dotenv import load_dotenv
from contextlib import AsyncExitStack
from mcp import ClientSession, StdioServerParameters
from mcp.client.stdio import stdio_client
import json
# 加载 .env 文件
load_dotenv()
class MCPClient:
def __init__(self):
"""初始化 MCP 客户端"""
self.exit_stack = AsyncExitStack()
self.api_key = os.getenv("API_KEY") # 读取 OpenAI API Key
self.base_url = os.getenv("BASE_URL") # 读取 BASE URL
self.model = os.getenv("MODEL") # 读取 model
if not self.api_key:
raise ValueError("未找到 API KEY. 请在 .env 文件中配置 API_KEY")
self.client = OpenAI(api_key=self.api_key, base_url=self.base_url)
self.sessions = {} # 存储多个服务端会话
self.tools_map = {} # 工具映射：工具名称 -> 服务端 ID
async def connect_to_server(self, server_id: str, server_script_path: str):
"""
连接到 MCP 服务器
:param server_id: 服务端标识符
:param server_script_path: 服务端脚本路径
"""
if server_id in self.sessions:
raise ValueError(f"服务端 {server_id} 已经连接")
is_python = server_script_path.endswith('.py')
is_js = server_script_path.endswith('.js')
if not (is_python or is_js):
raise ValueError("服务器脚本必须是 Python 或 JavaScript 文件")
command = "python" if is_python else "node"
server_params = StdioServerParameters(command=command,
args=[server_script_path],
env=None)
# 启动 MCP 服务器并建立通信
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))
await session.initialize()
self.sessions[server_id] = {"session": session, "stdio": stdio, "write": write}
print(f"已连接到 MCP 服务器: {server_id}")
# 更新工具映射
response = await session.list_tools()
for tool in response.tools:
self.tools_map[tool.name] = server_id
async def list_tools(self):
"""列出所有服务端的工具"""
if not self.sessions:
print("没有已连接的服务端")
return
print("已连接的服务端工具列表:")
for tool_name, server_id in self.tools_map.items():
print(f"工具: {tool_name}, 来源服务端: {server_id}")
async def process_query(self, query: str) -> str:
"""
调用大模型处理用户查询，并根据返回的 tools 列表调用对应工具。
支持多次工具调用，直到所有工具调用完成。
"""
messages = [{"role": "user", "content": query}]
# 构建统一的工具列表
available_tools = []
for tool_name, server_id in self.tools_map.items():
session = self.sessions[server_id]["session"]
response = await session.list_tools()
for tool in response.tools:
if tool.name == tool_name:
available_tools.append({
"type": "function",
"function": {
"name": tool.name,
"description": tool.description,
"input_schema": tool.inputSchema
}
})
print('整合的服务端工具列表:', available_tools)
# 循环处理工具调用
while True:
# 请求 OpenAI 模型处理
response = self.client.chat.completions.create(
model=self.model,
messages=messages,
tools=available_tools
)
# 处理返回的内容
content = response.choices[0]
if content.finish_reason == "tool_calls":
# 执行工具调用
for tool_call in content.message.tool_calls:
tool_name = tool_call.function.name
tool_args = json.loads(tool_call.function.arguments)
# 根据工具名称找到对应的服务端
server_id = self.tools_map.get(tool_name)
if not server_id:
raise ValueError(f"未找到工具 {tool_name} 对应的服务端")
session = self.sessions[server_id]["session"]
result = await session.call_tool(tool_name, tool_args)
print(f"
[Calling tool {tool_name} on server {server_id} with args {tool_args}]
")
# 将工具调用的结果添加到 messages 中
messages.append({
"role": "tool",
"content": result.content[0].text,
"tool_call_id": tool_call.id,
})
else:
# 如果没有工具调用，返回最终的回复
return content.message.content
async def chat_loop(self):
"""运行交互式聊天循环"""
print("MCP 客户端已启动！输入 'exit' 退出")
while True:
try:
query = input("问: ").strip()
if query.lower() == 'exit':
break
response = await self.process_query(query)
print(f"AI回复: {response}")
except Exception as e:
print(f"发生错误: {str(e)}")
async def clean(self):
"""清理所有资源"""
await self.exit_stack.aclose()
self.sessions.clear()
self.tools_map.clear()
async def main():
# 启动并初始化 MCP 客户端
client = MCPClient()
try:
# 连接多个 MCP 服务器
await client.connect_to_server("weather", '../../server/weather/weather.py')
await client.connect_to_server("filesystem", '../../server/filesystem/filesystem.py')
# 列出 MCP 服务器上的工具
await client.list_tools()
# 运行交互式聊天循环，处理用户对话
await client.chat_loop()
finally:
# 清理资源
await client.clean()
if __name__ == "__main__":
asyncio.run(main()) |
运行客户端：
新建一个终端：
| PowerShell
cd client
cd mcp-client
uv venv
.venvScriptsctivate
uv run client_tools.py |
测试
多工具测试建议使用Qwen 32B或DeepSeek V3，国内的其他小模型在工具使用上效果不堪入目。
在.env文件中配置：
| PowerShell
BASE_URL=https://api.siliconflow.cn/v1
MODEL=deepseek-ai/DeepSeek-V3
API_KEY="你的密钥" |
关闭重启启动一下，我们来测试一下客户端多工具使用的效果：
问：帮我查询广东省东莞市未来七天的天气预报，并且再G盘创建一个weather.txt文件，将查询到的天气预报内容写入文件中。
- 它首先调用了未来一个星期天气预报的工具：
- 获取到天气信息后，将结果返回给LLM，LLM获取数据后，还有一步没有做，紧接着就调用了文件创建的工具：
- 完成了文件创建的工具调用了，又返回结果给到LLM，LLM结合两次的返回最终返回：
🤭咱就是说挺棒的，就是硅基流动的DeepSeek V3的响应速度有点慢，每次响应都要一分钟左右，两个工具调用加最终的返回用了三分钟，你也赶紧试试吧。
下期预告：
- 优化客户端，通过配置文件引入本地服务器和云端服务器。
- 连接本地LLM大模型-构建完全本地的Agent。
