1.启动服务器 #

1.1 server.py #

# 导入logging模块，用于日志记录
import logging

# 从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": {}}


# 定义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)


# 定义运行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跟踪和自动清理")

    try:
        # 启动服务器，进入循环监听请求
        server.serve_forever()
    except KeyboardInterrupt:
        # 捕获键盘中断，记录服务器停止信息
        logger.info("HTTP服务器已停止")
    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()

2.客户端连接服务器 #

2.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, text/event-stream",
+               "MCP-Protocol-Version": self.protocol_version,
+           }
+       )

    # 生成消息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 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传输 ===")

    # 创建HTTP客户端实例
+   client = MCPClient("http://localhost:8000")

+   try:
        # 初始化客户端与服务器的连接
+       if not client.initialize():
+           print("初始化失败")
+           return

    # 捕获并打印异常
+   except Exception as e:
+       print(f"HTTP错误: {e}")


# 主函数
+def main():
+   """主函数"""
+   http_transport()


# 判断是否为主程序入口
+if __name__ == "__main__":
+   main()

2.2. server.py #

server.py

# 导入logging模块，用于日志记录
import logging

# 导入json模块，用于处理JSON数据
+import json

# 导入uuid模块，用于生成唯一会话ID
+import uuid

# 导入time模块，用于时间戳
+import time

# 从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 = {}

    # 处理初始化请求（旧实现，已被下方同名方法覆盖）
+   def handle_initialize(self, params, session_id: str = None):
+       """处理初始化请求"""
+       return {"capabilities": self.capabilities}

    # 生成唯一的会话ID
+   def _generate_session_id(self) -> str:
+       """生成唯一的会话ID"""
+       return str(uuid.uuid4())

    # 获取或创建会话
+   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 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
+       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或text/event-stream
+           accept_header = self.headers.get("Accept", "")
+           if (
+               "application/json" not in accept_header
+               and "text/event-stream" 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跟踪和自动清理")

    try:
        # 启动服务器，进入循环监听请求
        server.serve_forever()
    except KeyboardInterrupt:
        # 捕获键盘中断，记录服务器停止信息
        logger.info("HTTP服务器已停止")
    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()

3.获取工具列表 #

3.1. tools.py #

tools.py

# MCP工具定义

tools = [
    {
        "name": "calculate",
        "description": "执行数学计算",
        "inputSchema": {
            "type": "object",
            "properties": {
                "expression": {
                    "type": "string",
                    "description": "数学表达式",
                }
            },
            "required": ["expression"],
        },
    },
]

3.2. 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, text/event-stream",
                "MCP-Protocol-Version": self.protocol_version,
            }
        )

+   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)
+       if response and "result" in response:
+           tools = response["result"].get("tools", [])
+           logger.info(f"可用工具数量: {len(tools)}")
+           return tools
+       else:
+           logger.error("获取工具列表失败")
+           return []

+   def _send_message(self, message):
+       """通过HTTP发送消息"""
+       try:
            # 添加会话ID头（如果存在）
+           headers = {}
+           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")
+               if new_session_id and new_session_id != self.session_id:
+                   self.session_id = new_session_id
+                   logger.info(f"更新会话ID: {self.session_id}")

+               return response.json()
+           elif response.status_code == 202:
                # Accepted - 通知或响应
+               return None
+           elif response.status_code == 400:
+               logger.error(f"请求错误: {response.text}")
+               return None
+           elif response.status_code == 401:
+               logger.error("会话ID无效，需要重新初始化")
+               self.session_id = None
+               return None
+           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 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传输 ===")

    # 创建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]}")

    # 捕获并打印异常
    except Exception as e:
        print(f"HTTP错误: {e}")


# 主函数
def main():
    """主函数"""
    http_transport()


# 判断是否为主程序入口
if __name__ == "__main__":
    main()

3.3. server.py #

server.py

# 导入logging模块，用于日志记录
import logging

# 导入json模块，用于处理JSON数据
import json

# 导入uuid模块，用于生成唯一会话ID
import uuid

# 导入time模块，用于时间戳
import time
+from tools import tools

# 从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 = {}

+   def handle_tools_list(self, params=None, session_id: str = None):
+       """处理工具列表请求"""
+       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"""
        return str(uuid.uuid4())

    # 获取或创建会话
    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 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
        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或text/event-stream
            accept_header = self.headers.get("Accept", "")
            if (
                "application/json" not in accept_header
                and "text/event-stream" 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跟踪和自动清理")

    try:
        # 启动服务器，进入循环监听请求
        server.serve_forever()
    except KeyboardInterrupt:
        # 捕获键盘中断，记录服务器停止信息
        logger.info("HTTP服务器已停止")
    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()

4.调用工具列表 #

4.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, text/event-stream",
                "MCP-Protocol-Version": self.protocol_version,
            }
        )

    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)
        if response and "result" in response:
            tools = response["result"].get("tools", [])
            logger.info(f"可用工具数量: {len(tools)}")
            return tools
        else:
            logger.error("获取工具列表失败")
            return []

    def _send_message(self, message):
        """通过HTTP发送消息"""
        try:
            # 添加会话ID头（如果存在）
            headers = {}
            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")
                if new_session_id and new_session_id != self.session_id:
                    self.session_id = new_session_id
                    logger.info(f"更新会话ID: {self.session_id}")

                return response.json()
            elif response.status_code == 202:
                # Accepted - 通知或响应
                return None
            elif response.status_code == 400:
                logger.error(f"请求错误: {response.text}")
                return None
            elif response.status_code == 401:
                logger.error("会话ID无效，需要重新初始化")
                self.session_id = None
                return None
            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 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)
+       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传输 ===")

    # 创建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]}")

        # 调用calculate工具
+       result = client.call_tool("calculate", {"expression": "10 / 2 + 5"})
+       if result:
+           content = result.get("content", [])
+           if content:
+               print(f"计算结果: {content[0].get('text', '')}")

    # 捕获并打印异常
    except Exception as e:
        print(f"HTTP错误: {e}")


# 主函数
def main():
    """主函数"""
    http_transport()


# 判断是否为主程序入口
if __name__ == "__main__":
    main()

4.2. server.py #

server.py

# 导入logging模块，用于日志记录
import logging

# 导入json模块，用于处理JSON数据
import json

# 导入uuid模块，用于生成唯一会话ID
import uuid

# 导入time模块，用于时间戳
import time
from tools import tools

# 从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 = {}

    def handle_tools_list(self, params=None, session_id: str = None):
        """处理工具列表请求"""
        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"""
        return str(uuid.uuid4())

    # 获取或创建会话
    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 handle_tools_call(self, params, session_id: str = None):
+       """处理工具调用请求"""
+       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", {})

+           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)}"},
+                       }
+                   )
+           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
        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或text/event-stream
            accept_header = self.headers.get("Accept", "")
            if (
                "application/json" not in accept_header
                and "text/event-stream" 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跟踪和自动清理")

    try:
        # 启动服务器，进入循环监听请求
        server.serve_forever()
    except KeyboardInterrupt:
        # 捕获键盘中断，记录服务器停止信息
        logger.info("HTTP服务器已停止")
    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()

5.SSE传输 #

5.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, text/event-stream",
                "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}")

    # 打开SSE流连接方法
+   def open_sse_stream(self):
+       """打开SSE流连接"""
+       try:
            # 构造请求头
+           headers = {"Accept": "text/event-stream"}
            # 如果有会话ID则加入请求头
+           if self.session_id:
+               headers["Mcp-Session-Id"] = self.session_id

            # 发送GET请求，开启SSE流
+           response = self.session.get(
+               self.mcp_endpoint, headers=headers, stream=True, timeout=30
+           )

            # 如果响应状态码为200，表示连接成功
+           if response.status_code == 200:
+               logger.info("SSE流已打开")
+               return response
+           else:
                # 连接失败，记录错误
+               logger.error(f"SSE流打开失败: {response.status_code}")
+               return None

        # 捕获请求异常
+       except requests.exceptions.RequestException as e:
+           logger.error(f"SSE流连接错误: {e}")
+           return None

    # 获取当前会话信息方法
    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 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传输 ===")

    # 创建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]}")

        # 调用calculate工具
        result = client.call_tool("calculate", {"expression": "10 / 2 + 5"})
        if result:
            # 获取内容字段
            content = result.get("content", [])
            if content:
                print(f"计算结果: {content[0].get('text', '')}")
        # 打开SSE流并持续监听
+       sse_response = client.open_sse_stream()
+       if sse_response:
+           print(" SSE流连接成功")
+           try:
                # 持续监听SSE事件
+               for line in sse_response.iter_lines(decode_unicode=True):
                    # 处理event事件类型行
+                   if line.startswith("event: "):
+                       event_type = line[7:]  # 去掉 'event: ' 前缀
                    # 处理data数据行
+                   elif line.startswith("data: "):
+                       data_str = line[6:]  # 去掉 'data: ' 前缀
+                       try:
                            # 解析JSON数据
+                           data = json.loads(data_str)
                            # 处理connected事件
+                           if event_type == "connected":
+                               status = data.get("status")
+                               session_id = data.get("sessionId")
+                               print(f"🔗 连接状态: {status}, 会话ID: {session_id}")
                            # 处理time_update事件
+                           elif event_type == "time_update":
+                               server_time = data.get("server_time")
+                               timestamp = data.get("timestamp")
+                               session_id = data.get("sessionId")
+                               print(
+                                   f"🕐 服务器时间: {server_time} (时间戳: {timestamp}, 会话ID: {session_id})"
+                               )
                            # 处理未知事件类型
+                           else:
+                               print(f" 未知事件类型: {event_type}, 数据: {data}")
                        # 捕获JSON解析异常
+                       except json.JSONDecodeError:
+                           print(f" 数据解析错误: {data_str}")
                    # 空行表示事件结束，跳过
+                   elif line == "":
+                       continue
            # 捕获用户中断
+           except KeyboardInterrupt:
+               print("\n⏹️  用户停止监听")
+           finally:
                # 关闭SSE流
+               sse_response.close()
+               print("🔌 SSE流已关闭")

        # 最终会话状态
+       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()

5.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

# 导入datetime模块，用于处理日期和时间
+import datetime

# 从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 = {}

    # 获取所有活跃会话信息
+   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 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", {})

            # 如果工具名称为"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)}"},
                        }
                    )
            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)

    # 处理GET请求（建立SSE流）
+   def do_GET(self):
+       """处理GET请求（建立SSE流）"""
+       try:
            # 如果请求路径不是/mcp，返回404
+           if self.path != "/mcp":
+               self.send_error(404, "MCP endpoint not found")
+               return

            # 检查Accept头
+           accept_header = self.headers.get("Accept", "")
            # 如果Accept头不包含text/event-stream，返回405
+           if "text/event-stream" not in accept_header:
+               self.send_error(405, "Method not allowed")
+               return

            # 提取会话ID
+           session_id = self.headers.get("Mcp-Session-Id")
+           if session_id:
                # 验证会话是否存在
+               session_info = self.mcp_server.get_session_info(session_id)
+               if not session_info:
+                   self.send_error(401, "Invalid session ID")
+                   return
                # 记录使用现有会话
+               logger.info(f"SSE连接使用现有会话: {session_id}")
+           else:
                # 为SSE连接创建新会话
+               session_id = self.mcp_server._get_or_create_session(None)
                # 记录新会话
+               logger.info(f"SSE连接创建新会话: {session_id}")

            # 设置SSE头
+           self.send_response(200)
+           self.send_header("Content-Type", "text/event-stream")
+           self.send_header("Cache-Control", "no-cache")
+           self.send_header("Connection", "keep-alive")
+           self.send_header("MCP-Protocol-Version", self.mcp_server.protocol_version)
+           self.send_header("Mcp-Session-Id", session_id)  # 返回会话ID
+           self.end_headers()

            # 发送初始事件
+           self._send_sse_event(
+               "connected", {"status": "connected", "sessionId": session_id}
+           )
            # 记录SSE连接建立
+           logger.info(f"SSE连接已建立，会话ID: {session_id}，开始发送时间更新")

            # 定时发送服务器时间
+           try:
+               while True:
                    # 每5秒发送一次
+                   time.sleep(5)
                    # 获取当前服务器时间
+                   current_time = datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S")
                    # 发送时间更新事件
+                   self._send_sse_event(
+                       "time_update",
+                       {
+                           "server_time": current_time,
+                           "timestamp": time.time(),
+                           "sessionId": session_id,
+                       },
+                   )
                    # 记录发送时间
+                   logger.info(f"发送时间更新: {current_time}, 会话ID: {session_id}")
+           except (BrokenPipeError, ConnectionResetError):
                # 连接断开，记录日志
+               logger.info(f"SSE连接已断开，会话ID: {session_id}")
+           except Exception as e:
                # 发送事件出错，记录日志
+               logger.error(f"SSE事件发送错误: {e}, 会话ID: {session_id}")

+       except Exception as e:
            # 处理GET请求出错，记录日志
+           logger.error(f"GET处理错误: {e}")
            # 返回500错误
+           self.send_error(500, "Internal Server Error")

    # 发送SSE事件
+   def _send_sse_event(self, event_type: str, data):
+       """发送SSE事件"""
+       try:
            # 构造事件数据
+           event_data = f"event: {event_type}\n"
+           event_data += f"data: {json.dumps(data)}\n\n"
            # 写入事件到输出流
+           self.wfile.write(event_data.encode("utf-8"))
+           self.wfile.flush()
+       except Exception as e:
            # 发送事件出错，记录日志
+           logger.error(f"发送SSE事件错误: {e}")

    # 处理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
        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或text/event-stream
            accept_header = self.headers.get("Accept", "")
            if (
                "application/json" not in accept_header
                and "text/event-stream" 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()