1.实现取消功能 #
1.1. client.py #
client.py
# 导入json模块，用于处理JSON数据
import json
# 导入logging模块，用于日志记录
import logging
# 导入requests模块，用于发送HTTP请求
import requests
# 从urllib.parse模块导入urljoin，用于拼接URL
from urllib.parse import urljoin
# 配置日志级别为INFO
logging.basicConfig(level=logging.INFO)
# 获取当前模块的日志记录器
logger = logging.getLogger(__name__)
# 定义MCPClient类，表示MCP HTTP客户端
class MCPClient:
"""MCP HTTP客户端"""
# 初始化方法，设置服务器URL和相关属性
def __init__(self, server_url: str):
# 协议版本号
self.protocol_version = "2025-06-18"
# 客户端信息
self.client_info = {"name": "MCP客户端", "version": "1.0.0"}
# 服务器信息，初始化为None
self.server_info = None
# 服务器能力，初始化为None
self.capabilities = None
# 会话ID，初始化为None
self.session_id = None
# 消息ID，初始值为1
self.message_id = 1
# 去除服务器URL末尾的斜杠
self.server_url = server_url.rstrip("/")
# 拼接MCP接口的完整URL
self.mcp_endpoint = urljoin(self.server_url, "/mcp")
# 创建requests的Session对象
self.session = requests.Session()
# 设置Session的默认请求头
self.session.headers.update(
{
"Accept": "application/json",
"MCP-Protocol-Version": self.protocol_version,
}
)
# 重置会话方法
def reset_session(self):
"""重置会话"""
# 保存旧的会话ID
old_session_id = self.session_id
# 清空当前会话ID
self.session_id = None
# 记录会话重置信息
logger.info(f"会话已重置，原会话ID: {old_session_id}")
# 获取当前会话信息方法
def get_session_info(self):
"""获取当前会话信息"""
# 返回会话相关信息字典
return {
"session_id": self.session_id,
"server_url": self.server_url,
"protocol_version": self.protocol_version,
}
# 获取可用工具列表方法
def list_tools(self):
"""获取可用工具列表"""
# 构造请求消息
request = {
"jsonrpc": "2.0",
"id": self._generate_id(),
"method": "tools/list",
"params": {},
}
# 发送请求并获取响应
response = self._send_message(request)
# 如果响应存在且包含result字段
if response and "result" in response:
# 获取工具列表
tools = response["result"].get("tools", [])
logger.info(f"可用工具数量: {len(tools)}")
return tools
else:
# 获取失败，记录错误
logger.error("获取工具列表失败")
return []
# 通过HTTP发送消息的私有方法
def _send_message(self, message):
"""通过HTTP发送消息"""
try:
# 初始化请求头字典
headers = {}
# 如果有会话ID，则添加到请求头
if self.session_id:
headers["Mcp-Session-Id"] = self.session_id
logger.debug(f"使用会话ID: {self.session_id}")
# 发送POST请求
response = self.session.post(
self.mcp_endpoint, json=message, headers=headers, timeout=30
)
# 检查响应状态码
if response.status_code == 200:
# 检查响应头中是否有新的会话ID
new_session_id = response.headers.get("Mcp-Session-Id")
# 如果有新的会话ID且与当前不同，则更新
if new_session_id and new_session_id != self.session_id:
self.session_id = new_session_id
logger.info(f"更新会话ID: {self.session_id}")
# 返回响应的JSON内容
return response.json()
# 如果状态码为202，表示已接受但无内容
elif response.status_code == 202:
# Accepted - 通知或响应
return None
# 如果状态码为400，表示请求错误
elif response.status_code == 400:
logger.error(f"请求错误: {response.text}")
return None
# 如果状态码为401，表示会话ID无效
elif response.status_code == 401:
logger.error("会话ID无效，需要重新初始化")
self.session_id = None
return None
# 其他HTTP错误
else:
logger.error(f"HTTP错误: {response.status_code} - {response.text}")
return None
# 捕获请求异常
except requests.exceptions.RequestException as e:
logger.error(f"HTTP请求错误: {e}")
return None
# 生成消息ID的方法
def _generate_id(self) -> str:
"""生成消息ID"""
# 消息ID自增
self.message_id += 1
# 返回字符串类型的消息ID
return str(self.message_id)
# 发送消息到服务器的方法（重复定义，实际只会用后面这个）
def _send_message(self, message):
"""通过HTTP发送消息"""
try:
# 初始化请求头字典
headers = {}
# 如果有会话ID，则添加到请求头
if self.session_id:
headers["Mcp-Session-Id"] = self.session_id
logger.debug(f"使用会话ID: {self.session_id}")
# 发送POST请求到MCP端点
response = self.session.post(
self.mcp_endpoint, json=message, headers=headers, timeout=30
)
# 检查响应状态码
if response.status_code == 200:
# 从响应头获取新的会话ID
new_session_id = response.headers.get("Mcp-Session-Id")
# 如果有新的会话ID且与当前不同，则更新
if new_session_id and new_session_id != self.session_id:
self.session_id = new_session_id
logger.info(f"更新会话ID: {self.session_id}")
# 返回响应的JSON内容
return response.json()
# 如果状态码为202，表示已接受但无内容
elif response.status_code == 202:
# Accepted - 通知或响应
return None
# 如果状态码为400，表示请求错误
elif response.status_code == 400:
logger.error(f"请求错误: {response.text}")
return None
# 如果状态码为401，表示会话ID无效
elif response.status_code == 401:
logger.error("会话ID无效，需要重新初始化")
self.session_id = None
return None
# 其他HTTP错误
else:
logger.error(f"HTTP错误: {response.status_code} - {response.text}")
return None
# 捕获请求异常
except requests.exceptions.RequestException as e:
logger.error(f"HTTP请求错误: {e}")
return None
# 取消请求方法
+ def cancel_request(self, request_id: str, reason: str = "用户请求取消"):
+ """取消指定的请求"""
# 构造取消请求消息
+ request = {
+ "jsonrpc": "2.0",
+ "id": self._generate_id(),
+ "method": "cancellation",
+ "params": {"requestId": request_id, "reason": reason},
+ }
# 发送取消请求
+ response = self._send_message(request)
+ print(response)
+ if response and "result" in response:
+ result = response["result"]
+ status = result.get("status")
+ if status == "cancelled":
+ logger.info(f"请求 {request_id} 已成功取消")
+ return True
+ else:
+ logger.warning(f"取消请求失败: {status}")
+ return False
+ else:
+ logger.error("发送取消请求失败")
+ return False
# 调用长时间运行的工具方法
+ def call_long_running_tool(self, tool_name, arguments):
+ """调用长时间运行的工具"""
# 构造调用工具的请求消息
+ request = {
+ "jsonrpc": "2.0",
+ "id": self._generate_id(),
+ "method": "tools/call",
+ "params": {"calls": [{"name": tool_name, "arguments": arguments}]},
+ }
# 发送请求并获取响应
+ response = self._send_message(request)
+ if response and "result" in response:
+ calls = response["result"].get("calls", [])
+ if calls:
+ return calls[0]
+ else:
+ logger.error(f"调用工具 {tool_name} 失败")
+ return None
# 调用工具方法
def call_tool(self, tool_name, arguments):
"""调用工具"""
# 构造调用工具的请求消息
request = {
"jsonrpc": "2.0",
"id": self._generate_id(),
"method": "tools/call",
"params": {"calls": [{"name": tool_name, "arguments": arguments}]},
}
# 发送请求并获取响应
response = self._send_message(request)
# 如果响应存在且包含result字段
if response and "result" in response:
# 获取调用结果
calls = response["result"].get("calls", [])
if calls:
return calls[0]
else:
# 调用失败，记录错误
logger.error(f"调用工具 {tool_name} 失败")
return None
# 初始化与服务器连接的方法
def initialize(self) -> bool:
"""初始化与服务器的连接"""
# 构造初始化请求消息
request = {
"jsonrpc": "2.0",
"id": self._generate_id(),
"method": "initialize",
"params": {
"protocolVersion": self.protocol_version,
"capabilities": {},
"clientInfo": self.client_info,
},
}
# 发送初始化请求
response = self._send_message(request)
# 如果响应存在且包含result字段
if response and "result" in response:
result = response["result"]
# 获取服务器信息
self.server_info = result.get("serverInfo")
# 获取服务器能力
self.capabilities = result.get("capabilities")
# 提取并存储会话ID
if "sessionId" in result:
self.session_id = result["sessionId"]
logger.info(f"获取到会话ID: {self.session_id}")
logger.info(f"服务器初始化成功: {self.server_info}")
return True
else:
# 初始化失败，记录错误
logger.error("服务器初始化失败")
return False
# HTTP传输的函数
def http_transport():
"""HTTP传输"""
print("=== HTTP传输 ===")
client = MCPClient("http://localhost:8000")
try:
if not client.initialize():
print("初始化失败")
return
session_info = client.get_session_info()
print(f" 会话信息: {session_info}")
tools = client.list_tools()
print(f"可用工具: {[tool['name'] for tool in tools]}")
result = client.call_tool("calculate", {"expression": "10 / 2 + 5"})
if result:
content = result.get("content", [])
if content:
print(f"计算结果: {content[0].get('text', '')}")
# 演示请求取消功能
+ print("\n=== 演示请求取消功能 ===")
# 启动一个长时间运行的任务
+ print(" 启动长时间运行任务...")
+ result = client.call_long_running_tool(
+ "long_running_task", {"duration": 15, "task_id": "long_task_001"}
+ )
+ if result:
+ print(f"任务启动结果: {result}")
# 等待2秒后取消任务
+ import time
+ print("⏳ 等待2秒后取消任务...")
+ time.sleep(2)
# 取消任务
+ print(" 发送取消请求...")
+ if client.cancel_request("long_task_001", "演示取消功能"):
+ print(" 取消请求发送成功")
+ else:
+ print(" 取消请求发送失败")
+ else:
+ print(" 无法启动长时间运行任务")
final_session_info = client.get_session_info()
print(f"🏁 最终会话状态: {final_session_info}")
print("\n=== 会话重置 ===")
client.reset_session()
reset_session_info = client.get_session_info()
print(f"🔄 重置后会话状态: {reset_session_info}")
except Exception as e:
print(f"HTTP错误: {e}")
# 主函数
def main():
"""主函数"""
# 调用HTTP传输函数
http_transport()
# 判断是否为主程序入口
if __name__ == "__main__":
main()
1.2. server.py #
server.py
# 导入logging模块，用于日志记录
import logging
# 导入json模块，用于处理JSON数据
import json
# 导入uuid模块，用于生成唯一会话ID
import uuid
# 导入time模块，用于时间戳
import time
# 从tools模块导入tools对象
from tools import tools
# 导入threading模块，用于多线程处理
+import threading
# 从http.server模块导入HTTPServer和BaseHTTPRequestHandler类
from http.server import HTTPServer, BaseHTTPRequestHandler
# 配置日志记录级别为INFO
logging.basicConfig(level=logging.INFO)
# 获取当前模块的日志记录器
logger = logging.getLogger(__name__)
# 定义MCPServer类，作为MCP服务器的核心类
class MCPServer:
"""MCP服务器核心类"""
# 初始化方法
def __init__(self):
# 协议版本号
self.protocol_version = "2025-06-18"
# 服务器信息，包括名称和版本
self.server_info = {"name": "MCP服务器", "version": "1.0.0"}
# 服务器能力，包括工具、资源和提示
self.capabilities = {"tools": {}, "resources": {}, "prompts": {}}
# 会话管理字典
self.sessions = {}
# 活跃请求字典，用于跟踪可取消的请求
+ self.active_requests = {}
# 获取所有活跃会话信息
def get_all_sessions(self):
"""获取所有活跃会话信息"""
return self.sessions.copy()
# 处理工具列表请求
def handle_tools_list(self, params=None, session_id: str = None):
"""处理工具列表请求"""
# 如果有会话ID，则获取或创建会话
if session_id:
self._get_or_create_session(session_id)
# 返回工具列表
return {"tools": tools}
# 处理初始化请求（旧方法，已被覆盖）
def handle_initialize(self, params, session_id: str = None):
"""处理初始化请求"""
# 返回服务器能力
return {"capabilities": self.capabilities}
# 生成唯一的会话ID
def _generate_session_id(self) -> str:
"""生成唯一的会话ID"""
# 使用uuid4生成唯一ID
return str(uuid.uuid4())
# 获取会话信息
def get_session_info(self, session_id: str):
"""获取会话信息"""
# 如果会话ID存在于sessions字典中，返回会话信息
if session_id in self.sessions:
return self.sessions[session_id]
# 否则返回None
return None
# 获取或创建会话
def _get_or_create_session(self, session_id: str = None):
"""获取或创建会话"""
# 如果没有传入会话ID，则生成一个新的
if not session_id:
session_id = self._generate_session_id()
# 如果会话ID不存在于sessions字典中，则新建一个会话
if session_id not in self.sessions:
self.sessions[session_id] = {
"created_at": time.time(),
"last_activity": time.time(),
"message_count": 0,
"client_info": None,
}
# 记录新会话的创建
logger.info(f"创建新会话: {session_id}")
# 更新会话的最后活动时间
self.sessions[session_id]["last_activity"] = time.time()
# 增加消息计数
self.sessions[session_id]["message_count"] += 1
# 返回会话ID
return session_id
# 清理已完成的请求
+ def _cleanup_completed_request(self, request_id: str):
+ """清理已完成的请求"""
+ if request_id in self.active_requests:
+ del self.active_requests[request_id]
+ logger.info(f"清理已完成的请求: {request_id}")
# 检查请求是否已取消
+ def _is_request_cancelled(self, request_id: str):
+ """检查请求是否已被取消"""
+ if request_id in self.active_requests:
+ return self.active_requests[request_id]["cancelled"]
+ return False
# 取消请求
+ def _cancel_request(self, request_id: str, reason: str = None):
+ """取消指定的请求"""
+ if request_id in self.active_requests:
+ self.active_requests[request_id]["cancelled"] = True
+ self.active_requests[request_id]["cancellation_reason"] = reason
+ logger.info(f"请求已取消: {request_id}, 原因: {reason}")
+ return True
+ else:
+ logger.warning(f"尝试取消未知请求: {request_id}")
+ return False
# 注册活跃请求
+ def _register_active_request(
+ self, request_id: str, session_id: str, method: str, start_time: float
+ ):
+ """注册一个活跃请求"""
+ self.active_requests[request_id] = {
+ "session_id": session_id,
+ "method": method,
+ "start_time": start_time,
+ "cancelled": False,
+ "cancellation_reason": None,
+ }
+ logger.info(f"注册活跃请求: {request_id} (方法: {method}, 会话: {session_id})")
# 处理取消通知
+ def handle_cancellation(self, params, session_id: str = None):
+ """处理取消通知"""
+ request_id = params.get("requestId")
+ reason = params.get("reason", "用户请求取消")
+ logger.info(
+ f"收到取消请求: requestId={request_id}, reason={reason}, session_id={session_id}"
+ )
+ if not request_id:
+ logger.error("取消通知缺少requestId")
+ return None
# 验证请求是否属于当前会话
+ logger.info(f"当前活跃请求: {self.active_requests}")
+ if request_id in self.active_requests:
+ request_info = self.active_requests[request_id]
+ logger.info(f"找到活跃请求: {request_info}")
+ if request_info["session_id"] != session_id:
+ logger.warning(f"会话 {session_id} 尝试取消其他会话的请求 {request_id}")
+ return None
+ else:
+ logger.warning(f"请求ID {request_id} 不在活跃请求列表中")
+ logger.info(f"当前活跃请求: {list(self.active_requests.keys())}")
# 执行取消
+ if self._cancel_request(request_id, reason):
+ logger.info(f"成功取消请求 {request_id}: {reason}")
+ return {"status": "cancelled", "requestId": request_id, "reason": reason}
+ else:
+ logger.warning(f"无法取消请求 {request_id}: 请求不存在或已完成")
+ return {"status": "not_found", "requestId": request_id}
# 处理工具调用请求
def handle_tools_call(self, params, session_id: str = None):
"""处理工具调用请求"""
# 如果有会话ID，则获取或创建会话
if session_id:
self._get_or_create_session(session_id)
# 获取调用列表
calls = params.get("calls", [])
# 初始化结果列表
results = []
# 遍历每个调用
for call in calls:
# 获取工具名称
tool_name = call.get("name")
# 获取参数
arguments = call.get("arguments", {})
# 获取调用ID
+ call_id = call.get("id", str(uuid.uuid4()))
# 获取任务ID（如果提供的话）
+ task_id = arguments.get("task_id", call_id)
+ logger.info(
+ f"处理工具调用: {tool_name}, call_id: {call_id}, task_id: {task_id}"
+ )
# 注册活跃请求，使用任务ID作为请求ID
+ self._register_active_request(
+ task_id, session_id, f"tools/call:{tool_name}", time.time()
+ )
# 如果工具名称为"calculate"
if tool_name == "calculate":
# 获取表达式
expression = arguments.get("expression", "")
try:
# 计算表达式结果
result = eval(expression)
# 添加成功结果到结果列表
results.append(
{
"name": tool_name,
"content": [
{
"type": "text",
"text": f"计算结果: {expression} = {result}",
}
],
}
)
except Exception as e:
# 计算出错，添加错误信息
results.append(
{
"name": tool_name,
"isError": True,
"error": {"message": f"计算错误: {str(e)}"},
}
)
+ elif tool_name == "long_running_task":
# 长时间运行的任务，用于演示取消功能
+ duration = arguments.get("duration", 10)
+ logger.info(f"开始长时间运行任务: {task_id}, 预计耗时: {duration}秒")
+ logger.info(f"注册的请求ID: {task_id}")
# 立即返回任务已启动的响应
+ results.append(
+ {
+ "name": tool_name,
+ "content": [
+ {
+ "type": "text",
+ "text": f"任务 {task_id} 已启动，预计耗时: {duration}秒",
+ }
+ ],
+ }
+ )
# 在后台线程中运行任务
+ def run_task():
+ try:
+ start_time = time.time()
+ cancelled = False
+ logger.info(f"后台任务开始执行: {task_id}")
+ for i in range(duration):
# 检查是否被取消
+ if self._is_request_cancelled(task_id):
+ logger.info(f"任务 {task_id} 已被取消")
+ cancelled = True
+ break
# 每秒检查一次
+ time.sleep(1)
+ elapsed = time.time() - start_time
+ logger.info(
+ f"任务 {task_id} 进度: {i+1}/{duration} ({elapsed:.1f}s)"
+ )
# 如果没有被取消，记录任务完成
+ if not cancelled:
+ logger.info(f"任务 {task_id} 已完成")
+ except Exception as e:
+ logger.error(f"任务 {task_id} 执行出错: {e}")
+ finally:
# 延迟清理请求，给取消请求留出更多时间
+ logger.info(f"任务 {task_id} 准备清理，等待10秒...")
+ time.sleep(10)
+ self._cleanup_completed_request(task_id)
+ logger.info(f"任务 {task_id} 已清理")
# 启动后台线程
+ task_thread = threading.Thread(target=run_task)
+ task_thread.daemon = True
+ task_thread.start()
else:
# 未知工具，添加错误信息
results.append(
{
"name": tool_name,
"isError": True,
"error": {"message": f"未知工具: {tool_name}"},
}
)
# 返回所有调用的结果
return {"calls": results}
# 处理初始化请求（实际生效的方法，覆盖上面的方法）
def handle_initialize(self, params, session_id: str = None):
"""处理初始化请求"""
# 获取客户端信息
client_info = params.get("clientInfo", {})
# 获取或创建会话ID
session_id = self._get_or_create_session(session_id)
# 存储客户端信息到会话
self.sessions[session_id]["client_info"] = client_info
# 记录客户端初始化信息
logger.info(f"客户端初始化: {client_info}, 会话ID: {session_id}")
# 返回初始化结果，包括协议版本、能力、服务器信息和会话ID
return {
"protocolVersion": self.protocol_version,
"capabilities": self.capabilities,
"serverInfo": self.server_info,
"sessionId": session_id, # 返回会话ID给客户端
}
# 定义MCPHTTPHandler类，继承自BaseHTTPRequestHandler
class MCPHTTPHandler(BaseHTTPRequestHandler):
"""HTTP处理器，支持Streamable HTTP传输"""
# 初始化方法，接收mcp_server参数
def __init__(self, *args, mcp_server=None, **kwargs):
# 保存MCP服务器实例
self.mcp_server = mcp_server
# 调用父类的初始化方法
super().__init__(*args, **kwargs)
# 处理MCP消息
def _handle_mcp_message(self, message, session_id: str = None):
"""处理MCP消息"""
# 获取方法名
method = message.get("method")
# 获取参数
params = message.get("params", {})
# 获取消息ID
msg_id = message.get("id")
# 判断方法类型
if method == "initialize":
# 调用服务器的handle_initialize方法
result = self.mcp_server.handle_initialize(params, session_id)
# 如果有消息ID，返回带ID的结果
return {"id": msg_id, "result": result} if msg_id else None
elif method == "tools/list":
# 处理工具列表请求
result = self.mcp_server.handle_tools_list(params, session_id)
return {"id": msg_id, "result": result} if msg_id else None
elif method == "tools/call":
# 处理工具调用请求
result = self.mcp_server.handle_tools_call(params, session_id)
return {"id": msg_id, "result": result} if msg_id else None
+ elif method == "cancellation":
# 处理取消通知
+ result = self.mcp_server.handle_cancellation(params, session_id)
+ return {"id": msg_id, "result": result} if msg_id else None
else:
# 如果方法不存在，返回错误
if msg_id:
return {
"id": msg_id,
"error": {"code": -32601, "message": f"Method not found: {method}"},
}
# 没有ID则返回None
return None
# 处理POST请求（发送消息到服务器）
def do_POST(self):
"""处理POST请求（发送消息到服务器）"""
try:
# 检查请求路径是否为/mcp
if self.path != "/mcp":
self.send_error(404, "MCP endpoint not found")
return
# 检查Accept头是否包含application/json
accept_header = self.headers.get("Accept", "")
if "application/json" not in accept_header:
self.send_error(400, "Missing required Accept header")
return
# 检查MCP-Protocol-Version头是否匹配
protocol_version = self.headers.get("MCP-Protocol-Version")
if (
protocol_version
and protocol_version != self.mcp_server.protocol_version
):
self.send_error(
400, f"Unsupported protocol version: {protocol_version}"
)
return
# 获取会话ID
session_id = self.headers.get("Mcp-Session-Id")
# 读取请求体长度
content_length = int(self.headers.get("Content-Length", 0))
# 读取请求体内容
body = self.rfile.read(content_length)
try:
# 尝试解析JSON消息
message = json.loads(body.decode("utf-8"))
except json.JSONDecodeError:
# JSON解析失败
self.send_error(400, "Invalid JSON")
return
# 处理MCP消息
response = self._handle_mcp_message(message, session_id)
# 发送响应
if response:
# 发送200响应码
self.send_response(200)
# 设置响应头Content-Type为application/json
self.send_header("Content-Type", "application/json")
# 设置协议版本头
self.send_header(
"MCP-Protocol-Version", self.mcp_server.protocol_version
)
# 如果响应中包含会话ID，则添加到响应头
if "result" in response and "sessionId" in response["result"]:
self.send_header("Mcp-Session-Id", response["result"]["sessionId"])
# 结束响应头
self.end_headers()
# 写入JSON响应体
self.wfile.write(json.dumps(response).encode("utf-8"))
else:
# 没有响应内容，返回202 Accepted
self.send_response(202) # Accepted
self.end_headers()
except Exception as e:
# 记录错误日志
logger.error(f"POST处理错误: {e}")
# 返回500内部服务器错误
self.send_error(500, "Internal Server Error")
# 定义运行HTTP服务器的函数
def run_http_server(mcp_server: MCPServer, host: str = "localhost", port: int = 8000):
"""运行HTTP服务器"""
# 定义处理器工厂函数，用于传递mcp_server实例
def handler_factory(*args, **kwargs):
return MCPHTTPHandler(*args, mcp_server=mcp_server, **kwargs)
# 创建HTTPServer实例，绑定主机和端口
server = HTTPServer((host, port), handler_factory)
# 记录服务器启动信息
logger.info(f"HTTP服务器运行在 http://{host}:{port}/mcp")
# 记录会话管理功能启用信息
logger.info("会话管理功能已启用，支持会话ID跟踪和自动清理")
# 启动会话清理线程
import threading
# 定义会话清理函数
def cleanup_sessions():
while True:
try:
# 每5分钟清理一次
time.sleep(300)
# 调用会话清理方法
cleaned_count = mcp_server._cleanup_expired_sessions()
# 如果有清理的会话，记录日志
if cleaned_count > 0:
logger.info(f"清理了 {cleaned_count} 个过期会话")
# 记录当前活跃会话数
active_sessions = len(mcp_server.get_all_sessions())
logger.debug(f"当前活跃会话数: {active_sessions}")
except Exception as e:
# 清理出错，记录日志
logger.error(f"会话清理错误: {e}")
# 启动清理线程
cleanup_thread = threading.Thread(target=cleanup_sessions, daemon=True)
cleanup_thread.start()
logger.info("会话清理线程已启动")
try:
# 启动服务器，进入循环监听请求
server.serve_forever()
except KeyboardInterrupt:
# 捕获Ctrl+C，记录服务器停止
logger.info("HTTP服务器已停止")
logger.info("正在清理所有会话...")
# 清理所有会话
all_sessions = mcp_server.get_all_sessions()
for session_id in list(all_sessions.keys()):
del mcp_server.sessions[session_id]
logger.info(f"已清理 {len(all_sessions)} 个会话")
finally:
# 关闭服务器
server.server_close()
# 定义主函数
def main():
"""主函数"""
# 导入argparse模块，用于解析命令行参数
import argparse
# 创建ArgumentParser对象，设置描述信息
parser = argparse.ArgumentParser(description="MCP HTTP服务器")
# 添加--host参数，指定服务器主机，默认localhost
parser.add_argument(
"--host", default="localhost", help="HTTP服务器主机 (默认: localhost)"
)
# 添加--port参数，指定服务器端口，默认8000
parser.add_argument(
"--port", type=int, default=8000, help="HTTP服务器端口 (默认: 8000)"
)
# 解析命令行参数
args = parser.parse_args()
# 创建MCP服务器实例
mcp_server = MCPServer()
# 运行HTTP服务器
run_http_server(mcp_server, args.host, args.port)
# 判断是否为主程序入口
if __name__ == "__main__":
# 调用主函数
main()
1.3. tools.py #
tools.py
# MCP工具定义
tools = [
{
"name": "calculate",
"description": "执行数学计算",
"inputSchema": {
"type": "object",
"properties": {
"expression": {
"type": "string",
"description": "数学表达式",
}
},
"required": ["expression"],
},
},
+ {
+ "name": "long_running_task",
+ "description": "长时间运行的任务，用于演示取消功能",
+ "inputSchema": {
+ "type": "object",
+ "properties": {
+ "duration": {
+ "type": "integer",
+ "description": "任务持续时间（秒）",
+ "default": 10,
+ "minimum": 1,
+ "maximum": 60,
+ },
+ "task_id": {
+ "type": "string",
+ "description": "任务ID（可选）",
+ },
+ },
+ "required": ["duration"],
+ },
+ },
]