简介
该项目实现了一个入门的MCP Client和MCP Server交互的功能。 MCP Client中使用了React Agent模式来实现工具的调用。 MCP Server实现了两种启动方式即（stdio和sse），该例子中使用了sse的方式。
创建Agent
使用ReAct模式创建一个Agent，也可以使用langchain等框架实现，我这里没有使用AI框架。 在项目目录下创建‘react_agent.py’文件
import asyncio
import json
import os
import datetime
from openai import AsyncOpenAI
from mcp_client import MCPClient
from pprint import pprint
class ReActAgent:
def __init__(self, mcp_client: MCPClient):
# 从环境变量获取API密钥，如果不存在则使用默认值
# 修改base_url来使用不同的大模型供应商
# React模式一次使用一个工具
api_key = os.environ.get("API_KEY", "your llm api key")
self.llm = AsyncOpenAI(
api_key=api_key,
base_url="https://dashscope.aliyuncs.com/compatible-mode/v1",
)
self.mcp_client = mcp_client
self.messages = []
self.system_prompt = """
你是一个专业会议室预定AI助手，必须严格遵循以下规则运行：
【核心机制】
采用Thought→Action→Observation循环工作流：
1. 解析用户需求时，必须明确识别以下要素：
- 精确到分钟的时间段（自动补全日期格式）
- 预定目标（查询/预定）
2. 工具调用规范：
只能使用下列工具且每次只能调用1个：
- list_idle_meeting_rooms(必填参数：start_time, end_time)
- book_meeting_room(必填参数：room_id, start_time, end_time)
参数要求：
• 时间参数必须转为"YYYY-MM-DD HH:mm:ss"格式
• room_id必须从list_idle_meeting_rooms返回结果中选取
3. 执行流程强制要求：
(1) 预定操作前必须调用list_idle_meeting_rooms验证时段可用性
【输出规则】
1. 未完成时输出json格式：
{
"thought": "推理过程（必须包含时间转换逻辑）",
"action": {"name": "工具名", "args": {"参数1":"值1"}},
"observation": "工具返回原始结果"
}
2. 未完成标准：
- 未调用工具
- 未返回最终答案
- 明确提示预定失败
3. 完成标准：当且仅当满足以下条件时输出最终答案：
- book_meeting_room返回预定成功提示
- 包含有效的会议室ID
4. 最终答案json格式：
{
"final_answer": "预定成功：{room_id} ({start_time}至{end_time})"
}
【校验机制】
1. 时间参数三重验证：
(1) 格式正则校验：^\d{4}-\d{2}-\d{2} \d{2}:\d{2}:\d{2}$
(2) 时间逻辑：start_time < end_time
(3) 时段冲突检测（通过list_idle_meeting_rooms实现）
2. 错误处理：
- 参数缺失时精确指出缺失项，示例：
{"action":{"name":"request_params","args":{"missing_params":["start_time"]}}}
- 工具调用失败时返回原始错误信息
【终止条件】
1. 成功终止：输出含预定成功信息的final_answer
2. 失败终止：连续3次参数请求未获有效输入时返回：
{"final_answer": "ERR_MAX_RETRY: 超过最大重试次数"}
【错误处理与结果验证】
1. 必须严格检查工具返回结果中是否包含"失败"、"不存在"、"错误"等关键词
2. 当observation中包含任何失败信息时：
- 禁止输出final_answer
- 必须重新进行工具调用解决问题
3. 验证规则：
- 必须逐字分析observation返回内容
- 当且仅当observation明确包含"{room_id}预定成功"字样时才可输出最终答案
- 任何失败情况必须重试或告知用户失败原因
【列表解析规则】
1. list_idle_meeting_rooms返回结果必须按原格式提取，禁止自行编造会议室ID
2. 必须严格匹配返回的会议室ID格式，不得修改或推测
3. 如返回格式为["会议室1", "会议室2", "会议室3"]，则room_id必须是完全一致的"会议室1"、"会议室2"或"会议室3"
【强制验证机制】
1. 每次工具调用后必须执行：
if "失败" in observation or "不存在" in observation or "错误" in observation:
# 必须处理错误，不允许输出final_answer
# 必须尝试替代方案或告知用户
2. 禁止性规则：
- 严禁在任何失败场景下输出含"预定成功"的消息
- 严禁忽略工具返回的任何错误或警告信息
【强制纠错与重试机制】
1. 当工具返回错误时，必须执行以下步骤：
- 分析错误原因（例如："会议室不存在"）
- 重新查询可用会议室列表
- 从返回的实际可用会议室中选择
- 使用正确ID重试预定
2. 只有在确认预定成功后才能生成final_answer
【严格的上下文连贯性要求】
1. 每一步动作必须直接基于上一步observation结果，禁止忽略或歪曲
2. 必须逐字分析observation内容，并根据实际返回结果（而非臆想）执行下一步
3. 会议室ID必须严格从list_idle_meeting_rooms的结果中原样提取，禁止臆造或修改格式
4. 当observation包含"失败"、"不存在"等关键词时，下一步必须是纠正行动，而非宣告成功
【强制确认机制】
1. 在每次工具调用后，必须首先重复observation结果以证明理解
2. 思考过程必须包含对observation的明确分析，例如："根据观察到的'预定失败，R101不存在'，我需要重新选择正确的会议室ID"
3. 禁止在最终答案中使用与observation矛盾的信息
【结果验证关键规则】
1. 预定成功的唯一有效判断标准：book_meeting_room返回必须包含明确的"预定成功"字样
2. 会议室ID的有效来源：必须直接使用list_idle_meeting_rooms返回的会议室ID列表中的选项，不得添加任何前缀或后缀
3. 严禁从失败的预定结果中提取会议室ID作为最终答案
4. 当工具返回包含"失败"时，你必须尝试使用不同的会议室ID重新预定
"""
async def run(self, query):
try:
self.messages.append({"role": "system", "content": self.system_prompt})
self.messages.append({"role": "system", "content": f"当前时间：{datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S")}"})
self.messages.append({"role": "user", "content": query})
await self.mcp_client.connect_to_sse_server("http://127.0.0.1:8001/sse")
observation_history = []
for i in range(10): # 最大循环次数限制
print(f"第{i+1}次循环")
try:
response = await self.llm.chat.completions.create(
model="qwen-plus",
messages=self.messages,
response_format={"type": "json_object"},
)
message = response.choices[0].message
# 检查是否是最终答案
if message.content and "final_answer" in message.content:
try:
final_answer = json.loads(message.content)["final_answer"]
# 检查上一步observation是否包含失败信息，处理模型判断失误的情况
last_observation = observation_history[-1]["result"]
if (
"预定失败" in last_observation
or "不存在" in last_observation
or "错误" in last_observation
):
# 发现错误 - 强制继续对话
error_msg = {
"role": "user",
"content": f"警告！你错误的宣告了成功，工具返回的信息为：「{last_observation}」。请正确处理这个错误并重试，不要声称成功。",
}
self.messages.append(error_msg)
continue
return final_answer
except (json.JSONDecodeError, KeyError) as e:
print(f"解析final_answer时出错: {e}")
# 继续执行，尝试下一轮对话
# 解析并执行工具调用
# print("message.content", message.content)
action = json.loads(message.content)["action"]
tool_name = action["name"]
tool_args = action["args"]
# 调用工具
observation = await self.mcp_client.session.call_tool(
tool_name, tool_args
)
observation_text = str(observation.content[0].text)
print(
f"\n调用工具{tool_name}，参数{tool_args}，返回结果{observation_text}"
)
thought = json.loads(message.content)["thought"]
action["observation"] = observation_text
print(f"\n当前步骤{i+1}的reAct详情如下：")
print(f"thought: {thought}")
print(f"action: {tool_name}，参数{tool_args}")
print(f"observation: {observation_text}")
# 将思考步骤添加到对话上下文
self.messages.append(
{
"role": "assistant",
"content": json.dumps(
{
"thought": thought,
"action": action,
"observation": observation_text,
}
),
}
)
# 在工具调用后
observation_history.append(
{
"tool": tool_name,
"args": tool_args,
"result": observation_text,
}
)
except Exception as e:
print(f"API调用异常: {str(e)}")
continue
return "超过最大尝试次数，任务终止"
except Exception as e:
print(f"执行过程中发生异常: {str(e)}")
return f"执行失败: {str(e)}"
if __name__ == "__main__":
meetting_booking_mcp_client = MCPClient()
agent = ReActAgent(meetting_booking_mcp_client)
loop = asyncio.get_event_loop()
result = loop.run_until_complete(
agent.run(
"我想预定一个会议室，时间是明天上午9点到10点半,如果有多个空闲会议室，请随便选一个"
)
)
print(f"最终结果: {result}")
创建一个MCP Server
在项目目录下创建一个新的文件‘meeting_room.py’，粘贴下面的内容。 MCP Server使用sse方式的情况，需要在MCP Client启动前启动，所以可以使用‘python meeting_room.py sse 8001’启动服务。
import random
from mcp.server.sse import SseServerTransport
from mcp import types
from mcp.server.lowlevel import Server
import logging
from starlette.applications import Starlette
from starlette.routing import Mount, Route
import uvicorn
app = Server("meeting_room")
async def list_idle_meeting_rooms(start_time: str, end_time: str) -> list[types.TextContent]:
return ["会议室1", "会议室2", "会议室3"]
async def book_meeting_room(room_id: str, start_time: str, end_time: str) -> list[types.TextContent]:
# 随机返回预定成功或失败
print(f"预定会议室{room_id}，开始时间{start_time}，结束时间{end_time}")
if room_id in ["会议室1", "会议室2", "会议室3"]:
return f"{room_id}预定成功" if random.random() < 0.5 else f"{room_id}预定失败，请重新选择会议室"
else:
return f"预定失败，{room_id}不存在,请重新选择会议室"
@app.list_tools()
async def list_tools() -> list[types.Tool]:
return [
types.Tool(
name="list_idle_meeting_rooms",
description="list idle meeting rooms",
inputSchema={
"type": "object",
"properties": {
"start_time": {
"type": "string",
"description": "开始时间"
},
"end_time": {
"type": "string",
"description": "结束时间"
}
},
"required": ["start_time", "end_time"]
}
),
types.Tool(
name="book_meeting_room",
description="book meeting room",
inputSchema={
"type": "object",
"properties": {
"room_id": {
"type": "string",
"description": "会议室ID"
},
"start_time": {
"type": "string",
"description": "开始时间"
},
"end_time": {
"type": "string",
"description": "结束时间"
}
},
"required": ["room_id", "start_time", "end_time"]
}
)
]
@app.call_tool()
async def call_tool(
name: str,
arguments: dict
) -> list[types.TextContent | types.ImageContent | types.EmbeddedResource]:
if name == "list_idle_meeting_rooms":
start_time = arguments["start_time"]
end_time = arguments["end_time"]
result = await list_idle_meeting_rooms(start_time, end_time)
return [types.TextContent(type="text", text=str(result))]
elif name == "book_meeting_room":
print(arguments)
room_id = arguments["room_id"]
start_time = arguments["start_time"]
end_time = arguments["end_time"]
result = await book_meeting_room(room_id, start_time, end_time)
return [types.TextContent(type="text", text=str(result))]
raise ValueError(f"Tool not found: {name}")
if __name__ == "__main__":
# Run the FastAPI app with uvicorn
import sys
# 默认使用标准输入输出传输
transport = "stdio"
port = 8000
# 检查命令行参数
if len(sys.argv) > 1:
if sys.argv[1] == "sse":
transport = "sse"
if len(sys.argv) > 2:
try:
port = int(sys.argv[2])
except ValueError:
pass
if transport == "sse":
# 配置日志
logging.basicConfig(level=logging.INFO,
format='%(asctime)s - %(name)s - %(levelname)s - %(message)s')
logger = logging.getLogger("weather-mcp")
# 配置 SSE 传输，增加超时和重试参数
sse = SseServerTransport(
"/message/",
)
async def handle_sse(request):
try:
logger.info(f"建立新的SSE连接: {request.client}")
async with sse.connect_sse(request.scope, request.receive, request._send) as streams:
try:
await app.run(streams[0], streams[1], app.create_initialization_options())
except Exception as e:
logger.error(f"处理SSE连接时出错: {str(e)}")
import traceback
logger.error(traceback.format_exc())
except Exception as e:
logger.error(f"建立SSE连接时出错: {str(e)}")
import traceback
logger.error(traceback.format_exc())
return
starlette_app = Starlette(
debug=True,
routes=[
Route("/sse", endpoint=handle_sse),
Mount("/message/", app=sse.handle_post_message),
]
)
print(f"在端口{port}上启动MCP服务器，使用SSE传输")
print(f"连接URL: http://127.0.0.1:{port}/sse")
uvicorn.run(starlette_app, host="0.0.0.0", port=port, log_level="info")
else:
from mcp.server.stdio import stdio_server
import anyio
async def run_stdio():
async with stdio_server() as streams:
await app.run(streams[0], streams[1], app.create_initialization_options())
print("使用标准输入输出传输启动MCP服务器")
anyio.run(run_stdio)
创建一个MCP Server管理类
在项目目录创建‘mcp_client.py’文件
from typing import Optional
from contextlib import AsyncExitStack
from mcp.client.sse import sse_client
from mcp import ClientSession, StdioServerParameters
from mcp.client.stdio import stdio_client
class MCPClient:
def __init__(self):
self.session: Optional[ClientSession] = None
self.exit_stack = AsyncExitStack()
async def connect_to_server(self, server_script_path: str, server_interpreter_path: str):
"""Connect to an MCP server
Args:
server_script_path: Path to the server script (.py or .js)
"""
is_python = server_script_path.endswith(".py")
is_js = server_script_path.endswith(".js")
if not (is_python or is_js):
raise ValueError("Server script must be a .py or .js file")
command = "python" if is_python else "node"
command = server_interpreter_path
server_params = StdioServerParameters(
command=command, args=[server_script_path], env=None
)
stdio_transport = await self.exit_stack.enter_async_context(
stdio_client(server_params)
)
self.stdio, self.write = stdio_transport
self.session = await self.exit_stack.enter_async_context(
ClientSession(self.stdio, self.write)
)
await self.session.initialize()
# 列出可用的工具
response = await self.session.list_tools()
tools = response.tools
print("\nConnected to server with tools:", [tool.name for tool in tools])
async def connect_to_sse_server(self, server_url: str):
"""Connect to an MCP server
Args:
server_script_path: Path to the server script (.py or .js)
"""
self._streams_context = sse_client(url=server_url)
streams = await self._streams_context.__aenter__()
self._session_context = ClientSession(*streams)
self.session = await self._session_context.__aenter__()
await self.session.initialize()
# 列出可用的工具
response = await self.session.list_tools()
tools = response.tools
print("\nConnected to server with tools:", [tool.name for tool in tools])
启动MCP Client
使用命令启动agent，并从控制台观察输出。
python react_agent.py
控制台输出
第1次循环
调用工具list_idle_meeting_rooms，参数{'start_time': '2025-04-10 09:00:00', 'end_time': '2025-04-10 10:30:00'}，返回结果['会议室1', '会议室2', '会议室3']
当前步骤1的reAct详情如下：
thought: 用户想预定明天上午9点到10点半的会议室。首先需要将时间转换为正确的格式：start_time='2025-04-10 09:00:00'，end_time='2025-04-10 10:30:00'。然后调用list_idle_meeting_rooms 来查询这个时间段内空闲的会议室。
action: list_idle_meeting_rooms，参数{'start_time': '2025-04-10 09:00:00', 'end_time': '2025-04-10 10:30:00'}
observation: ['会议室1', '会议室2', '会议室3']
第2次循环
调用工具book_meeting_room，参数{'room_id': '会议室1', 'start_time': '2025-04-10 09:00:00', 'end_time': '2025-04-10 10:30:00'}，返回结果会议室1预定成功
当前步骤2的reAct详情如下：
thought: 查询结果显示有三个会议室可用：['会议室1', '会议室2', '会议室3']。用户要求随意选择一个，我将选择'会议室1'进行预定。现在需要调用book_meeting_room工具来完成预定操作。
action: book_meeting_room，参数{'room_id': '会议室1', 'start_time': '2025-04-10 09:00:00', 'end_time': '2025-04-10 10:30:00'}
observation: 会议室1预定成功
第3次循环
最终结果: 预定成功：会议室1 (2025-04-10 09:00:00至2025-04-10 10:30:00)