9.2.2 单机MCP 服务端的搭建
MCP 的作用是搭建工具函数与大模型应用的桥梁，我们首先完成工具函数类的设置，在具体使用上，读者需要安装MCP 专用的Python 包，代码如下：
pip install mcp
在本小节的示例中，我们需要实现多个可被MCP 调用的函数，并使用mcp.tool() 进行修饰，代码如下：
'''
MCP 服务端程序
提供 stdio 协议与 MCP 客户端进行通信
'''
import datetime
from mcp.server.fastmcp import FastMCP
mcp = FastMCP("DemoServer") # 创建服务实例
@mcp.tool()
def add_numbers(a: int, b: int) -> int:
"""
计算两个整数的和。
Args:
a (int): 第一个整数，例如： 5
b (int): 第二个整数，例如： 3
Returns:
int: 两数之和，例如： 8
"""
return a + b
@mcp.tool()
def multiply_numbers(a: int, b: int) -> int:
"""
计算两个整数的乘积。
Args:
a (int): 被乘数，例如： 4
b (int): 乘数，例如： 7
Returns:
int: 两数乘积，例如： 28
"""
return a * b
@mcp.tool()
def date_time() -> str:
"""
获取当前系统时间（基于服务器时区）。
Returns:
str: 格式为 YYYY-MM-DD HH:MM:SS 的时间字符串，例如： 2024-05-01 15:30:00
"""
return datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S")
@mcp.tool()
def get_weather(city_name: str) -> dict:
"""
获取指定城市的模拟天气信息（当前为静态数据）。
Args:
city_name (str): 城市名称（中文或拼音），例如： shanghai 或 上海
Returns:
dict: 包含城市和温度的字典，例如： {' 城市 ': ' 上海 ', ' 温度 ': '23 ℃ '}
"""
return {
' 城市 ': city_name,
' 温度 ': "23 ℃ "
}
if __name__ == "__main__":
mcp.run(transport="stdio") # 使用标准输入 / 输出协议运行服务
在上面代码中，我们定义了多个函数，并通过文本对函数作用进行说明。其中，mcp.tool() 的作用是对函数进行注册，将函数转换为可供MCP 框架使用的工具。
9.2.3 单机MCP 客户端的搭建 与使用
本小节将完成MCP 客户端的搭建。这里我们的目标是使用Qwen3 来完成工具的调用，首先需要完成Qwen3 的提示词设置与输出，代码如下：
from openai import OpenAI
from mcp import ClientSession, StdioServerParameters # 从 mcp 模块导入 ClientSession 和 StdioServerParameters 类
from mcp.client.stdio import stdio_client # 从 mcp.client.stdio 模块导入 stdio_client 函数
import sys # 导入 sys 模块，用于处理命令行参数
import json # 导入 json 模块，用于处理 JSON 数据
# 定义一个函数，用于转换 tools 工具中的 JSON 数据
def transform_json(tools):
s = "MCP 服务器提供的工具如下 :"
for tool in tools: # 遍历工具列表
s = s + f"""
tool_name: {tool.name},
tool_description: {tool.description},
- input title: {tool.inputSchema['title']},
- input properties:"{tool.inputSchema['properties']},
"""
return s
def ask_llm(question, tools_list):
# TODO: 实现与 LLM 的交互逻辑
# 这里只是一个示例，在实际应用中需要根据具体情况进行实现
system_prompt = tools_list + '\n 根据以上描述，用户要求 :%s ，请生成一个工具调用命令，要求以 JSON 格式输出 {"tool": 工具名 ,"tool_input": 参数字典 } ，只输出 JSON ，不要输出其他内容 ' % (
question)
client = OpenAI(
api_key="sk-17f687f6c5ac4647a9d9f649598a7cfe",
base_url="https://dashscope.aliyuncs.com/compatible-mode/v1",
)
response = client.chat.completions.create(
model="qwen-plus",
messages=[
{"role": "system", "content": system_prompt},
{"role": "user", "content": "Hello"},
],
max_tokens=1024,
temperature=0.99,
stream=False
)
generated_text = response.choices[0].message.content # 提取文本内容
return generated_text, client # 返回文本而非 response 对象
def llm_Qwen3(question,client):
response = client.chat.completions.create(
model="qwen-plus",
messages=[
{"role": "user", "content": question},
],
max_tokens=1024,
temperature=0.99,
stream=False
)
generated_text = response.choices[0].message.content # 提取文本内容
return generated_text
在上面代码中，定义了一个函数，用于转换tools 工具中的JSON 数据，而Qwen3 的作用是对输入的内容和函数进行匹配，并完成函数的调用。需要注意，我们使用system_prompt 对Qwen3 的作用以及输出格式做出规范，并且在system_prompt 中添加了函数的用法和说明。
完整的MCP 客户端搭建与使用代码如下：
from openai import OpenAI
from mcp import ClientSession, StdioServerParameters # 从 mcp 模块导入 ClientSession 和 StdioServerParameters 类
from mcp.client.stdio import stdio_client # 从 mcp.client.stdio 模块导入 stdio_client 函数
import sys # 导入 sys 模块，用于处理命令行参数
import json # 导入 json 模块，用于处理 JSON 数据
# 定义一个函数，用于转换 tools 工具中的 JSON 数据
def transform_json(tools):
s = "MCP 服务器提供的工具如下 :"
for tool in tools: # 遍历工具列表
s = s + f"""
tool_name: {tool.name},
tool_description: {tool.description},
- input title: {tool.inputSchema['title']},
- input properties:"{tool.inputSchema['properties']},
"""
return s
def ask_llm(question, tools_list):
# TODO: 实现与 LLM 的交互逻辑
# 这里只是一个示例，实际应用中需要根据具体情况进行实现
system_prompt = tools_list + '\n 根据以上描述，用户要求 :%s ，请生成一个工具调用命令，要求以 JSON 格式输出 {"tool": 工具名 ,"tool_input": 参数字典 } ，只输出 JSON ，不要输出其他内容 ' % (
question)
client = OpenAI(
api_key="sk-17f687f6c5ac4647a9d9f649598a7cfe",
base_url="https://dashscope.aliyuncs.com/compatible-mode/v1",
)
response = client.chat.completions.create(
model="qwen-plus",
messages=[
{"role": "system", "content": system_prompt},
{"role": "user", "content": "Hello"},
],
max_tokens=1024,
temperature=0.99,
stream=False
)
generated_text = response.choices[0].message.content # 提取文本内容
return generated_text, client # 返回文本而非 response 对象
def llm_Qwen3(question,client):
response = client.chat.completions.create(
model="qwen-plus",
messages=[
{"role": "user", "content": question},
],
max_tokens=1024,
temperature=0.99,
stream=False
)
generated_text = response.choices[0].message.content # 提取文本内容
return generated_text
import os
server_script_path = os.path.join(os.path.dirname(__file__), "mcp_tool_and_server.py")
# 定义服务器参数
server_params = StdioServerParameters(
command="python", # 运行命令
args=[server_script_path], # 服务器脚本路径
env=None # 可选的环境变量
)
async def run(question = " 你是谁 ?"):
# 建立连接
async with stdio_client(server_params) as (read, write):
async with ClientSession(read, write) as session:
# 初始化连接
await session.initialize()
# 列出可用工具
tools = await session.list_tools()
#print(" 可用工具 :", tools.tools)
s = transform_json(tools.tools) +"\n"
#print(s)
response, client = ask_llm(question,s)
# 清理 Markdown 标记
response = response.strip().replace('```json', '').replace('```', '').strip()
#print(response)
mtools = json.loads(response) # 现在 response 是纯 JSON 字符串
print("mtools: --> ",mtools)
if 'tool' in mtools:
tool_name = mtools['tool']
tool_input = mtools['tool_input']
# 调用工具
print(" 调用工具 :", tool_name, tool_input)
ret = await session.call_tool(tool_name, tool_input)
if ret:
try:
r = json.loads(ret.content[0].text)
except:
r = ret.content[0].text
print(" 工具返回结果 :", r)
questions = f" 用户的问题是 {question}, 根据 {tool_name} 的返回结果为： {r} ，根据以上信息 , 回答问题。 "
r = llm_Qwen3(questions,client)
print(r)
else:
r = llm_Qwen3(question,client)
print(r)
if __name__ == "__main__":
import asyncio
questions = [' 计算一下 356*125', '25+38 是多少 ?', ' 现在的时间是什么时候 ?', 'Shanghai 天气怎么样 ?',' 请给我讲一个笑话 ']
for question in questions:
asyncio.run(run(question))
上面代码的核心功能是通过一个本地工具服务器和LLM 结合，根据用户问题动态调用工具并生成答案。以下是代码的详细解析：
（1 ）模块引入：
from openai import OpenAI ：虽然命名为OpenAI ，但后续代码实际上是通过自定义类与Qwen3 的API 交互，这里可能是为了兼容或扩展性而命名。
from mcp import ClientSession, StdioServerParameters 和from mcp.client.stdio import stdio_client ：用于与MCP 服务器交互，ClientSession 管理会话，StdioServerParameters 配置服务器参数，stdio_client 用于创建客户端连接。
import sys, json ：sys 用于处理命令行参数（尽管代码中未直接使用），json 用于解析和生成JSON 数据。
import os ：用于处理文件路径，确定MCP 服务器脚本的位置。
（2 ）工具描述转换函数：
transform_json(tools) ：将MCP 服务器提供的工具列表转换为人类可读的字符串格式，包括工具名称、描述、输入标题和属性。这有助于将工具信息传递给LLM 以生成调用命令。
（3 ）LLM 交互函数：
ask_llm(question, tools_list) ：构建系统提示词，将工具描述和用户问题结合，生成一个JSON 格式的工具调用命令。通过Qwen3 API 发送请求，并返回生成的文本和客户端对象。
llm_Qwen3(question, client) ：直接通过客户端对象发送用户问题到LLM ，并返回生成的文本。
（4 ）工具调用与结果处理：
在run 异步函数中，首先通过stdio_client 和ClientSession 与MCP 服务器建立连接。
列出可用工具，并将其转换为可读字符串。
调用ask_llm 生成工具调用命令，解析返回的JSON 字符串以获取工具名称和输入参数。
如果成功解析出工具信息，则调用相应工具并处理返回结果。如果工具返回JSON 格式的结果，则尝试解析；否则，直接打印文本结果。
根据工具返回结果或原始问题，再次调用LLM 以生成最终答案。
最后使用asyncio.run(run(question)) 逐个处理每个问题。注意，这里每次处理一个问题都会重新建立与MCP 服务器的连接以及与LLM 的交互，这在实际应用中可能不是最高效的方式，但便于演示和测试。
=========================================
本文节选自《AI Agent智能体与MCP开发实践：基于Qwen3大模型》，获得出版社和作者授权发布。