1.服务器写入日志 #

1.1. client.py #

client.py

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

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

# 导入requests模块，用于发送HTTP请求
import requests

# 从urllib.parse模块导入urljoin，用于拼接URL
from urllib.parse import urljoin

# 配置日志级别为INFO
logging.basicConfig(level=logging.INFO)
# 获取当前模块的日志记录器
logger = logging.getLogger(__name__)


# 定义SSE监听器函数，负责持续监听服务器推送的事件
def start_sse_listerner(client):
    # 打开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})"
                            )
                        # 处理日志消息事件
+                       elif event_type == "log_message":
+                           level = data.get("level", "info")
+                           logger_name = data.get("logger", "unknown")
+                           log_data = data.get("data", {})
+                           message = log_data.get("message", "无消息")
+                           timestamp = log_data.get("server_time", "未知时间")

                            # 根据日志级别选择不同的显示符号和颜色
+                           level_icons = {
+                               "debug": "🔍",
+                               "info": "ℹ️",
+                               "notice": "📢",
+                               "warning": "⚠️",
+                               "error": "",
+                               "critical": "🚨",
+                               "alert": "🚨",
+                               "emergency": "💥",
+                           }

+                           icon = level_icons.get(level, "📝")
+                           print(f"{icon} [{level.upper()}] [{logger_name}] {message}")
+                           print(f"   📅 时间: {timestamp}")

                            # 显示额外的日志数据
+                           for key, value in log_data.items():
+                               if key not in ["message", "server_time"]:
+                                   print(f"   📊 {key}: {value}")
+                           print()  # 空行分隔
                        # 处理未知事件类型
                        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流已关闭")


# 定义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
        # 启动SSE监听线程
        sse_listerner_thread = threading.Thread(
            target=start_sse_listerner, args=(client,), daemon=True
        )
        sse_listerner_thread.start()
        # 显示会话信息
        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', '')}")

        # 最终会话状态
        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}")
+       sse_listerner_thread.join()
    # 捕获并打印异常
    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

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

# 从http.server模块导入ThreadingHTTPServer和BaseHTTPRequestHandler类
from http.server import ThreadingHTTPServer, 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": {}, "logging": {}}
        # 会话管理字典
        self.sessions = {}
        # 全局日志级别，默认为info
+       self.global_log_level = "info"
        # 日志级别映射
+       self.log_levels = {
+           "debug": 0,
+           "info": 1,
+           "notice": 2,
+           "warning": 3,
+           "error": 4,
+           "critical": 5,
+           "alert": 6,
+           "emergency": 7,
+       }

    # 发送日志消息
+   def send_log_message(
+       self, level: str, message: str, logger_name: str = None, data: dict = None
+   ):
+       """发送日志消息"""
        # 检查日志级别是否满足全局设置
+       if self.log_levels.get(level, 0) < self.log_levels.get(
+           self.global_log_level, 1
+       ):
+           return

        # 构造日志消息
+       log_data = {
+           "message": message,
+           "timestamp": time.time(),
+           "server_time": datetime.datetime.now().isoformat(),
+       }

        # 如果有额外数据，合并到日志数据中
+       if data:
+           log_data.update(data)

        # 记录到本地日志
+       log_method = getattr(logger, level, logger.info)
+       log_method(f"[{logger_name or 'server'}] {message}")

        # 返回日志消息结构（用于发送给客户端）
+       return {"level": level, "logger": logger_name or "server", "data": log_data}

    # 获取所有活跃会话信息
    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 _cleanup_expired_sessions(self, max_age: int = 3600):
        """清理过期会话（默认1小时）"""
        current_time = time.time()
        expired_sessions = []

        for session_id, session_data in self.sessions.items():
            if current_time - session_data["last_activity"] > max_age:
                expired_sessions.append(session_id)

        for session_id in expired_sessions:
            del self.sessions[session_id]
            logger.info(f"清理过期会话: {session_id}")

        return len(expired_sessions)

    # 获取或创建会话
    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)

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

    # 处理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}，开始发送时间更新")
            # 发送一些示例日志消息
+           self._send_log_event(
+               self.mcp_server.send_log_message(
+                   "info",
+                   "SSE连接已建立",
+                   "sse",
+                   {"session_id": session_id, "connection_type": "SSE"},
+               )
+           )
            # 定时发送服务器时间
            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)

    # 创建ThreadingHTTPServer实例，绑定主机和端口
    server = ThreadingHTTPServer((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()

2.设置日志级别 #

2.1. client.py #

client.py

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

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

# 导入requests模块，用于发送HTTP请求
import requests

# 从urllib.parse模块导入urljoin，用于拼接URL
from urllib.parse import urljoin

# 配置日志级别为INFO
logging.basicConfig(level=logging.INFO)
# 获取当前模块的日志记录器
logger = logging.getLogger(__name__)


# 定义SSE监听器函数，负责持续监听服务器推送的事件
def start_sse_listerner(client):
    # 打开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})"
                            )
                        # 处理日志消息事件
                        elif event_type == "log_message":
                            level = data.get("level", "info")
                            logger_name = data.get("logger", "unknown")
                            log_data = data.get("data", {})
                            message = log_data.get("message", "无消息")
                            timestamp = log_data.get("server_time", "未知时间")

                            # 根据日志级别选择不同的显示符号和颜色
                            level_icons = {
                                "debug": "🔍",
                                "info": "ℹ️",
                                "notice": "📢",
                                "warning": "⚠️",
                                "error": "",
                                "critical": "🚨",
                                "alert": "🚨",
                                "emergency": "💥",
                            }

                            icon = level_icons.get(level, "📝")
                            print(f"{icon} [{level.upper()}] [{logger_name}] {message}")
                            print(f"   📅 时间: {timestamp}")

                            # 显示额外的日志数据
                            for key, value in log_data.items():
                                if key not in ["message", "server_time"]:
                                    print(f"   📊 {key}: {value}")
                            print()  # 空行分隔
                        # 处理未知事件类型
                        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流已关闭")


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

    # 设置日志级别
+   def set_log_level(self, level: str) -> bool:
+       """设置服务器日志级别"""
+       request = {
+           "jsonrpc": "2.0",
+           "id": self._generate_id(),
+           "method": "logging/setLevel",
+           "params": {"level": level},
+       }

+       response = self._send_message(request)
+       if response and "result" in response:
+           result = response["result"]
+           if "success" in result and result["success"]:
+               logger.info(f"日志级别已设置为: {level}")
+               return True
+           elif "error" in result:
+               logger.error(f"设置日志级别失败: {result['error']['message']}")
+       else:
+           logger.error("设置日志级别失败")
+       return False

    # 打开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}")
            # 检查是否支持日志功能
+           if "logging" in self.capabilities:
+               logger.info("服务器支持日志功能")
                # 设置默认日志级别为info
+               self.set_log_level("info")
+           else:
+               logger.warning("服务器不支持日志功能")
            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
        # 启动SSE监听线程
        sse_listerner_thread = threading.Thread(
            target=start_sse_listerner, args=(client,), daemon=True
        )
        sse_listerner_thread.start()
        # 显示会话信息
        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', '')}")

        # 最终会话状态
        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}")
        sse_listerner_thread.join()
    # 捕获并打印异常
    except Exception as e:
        print(f"HTTP错误: {e}")


# 主函数
def main():
    """主函数"""
    # 调用HTTP传输函数
    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

# 从tools模块导入tools对象
from tools import tools

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

# 从http.server模块导入ThreadingHTTPServer和BaseHTTPRequestHandler类
from http.server import ThreadingHTTPServer, 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": {}, "logging": {}}
        # 会话管理字典
        self.sessions = {}
        # 全局日志级别，默认为info
        self.global_log_level = "info"
        # 日志级别映射
        self.log_levels = {
            "debug": 0,
            "info": 1,
            "notice": 2,
            "warning": 3,
            "error": 4,
            "critical": 5,
            "alert": 6,
            "emergency": 7,
        }

    # 发送日志消息
    def send_log_message(
        self, level: str, message: str, logger_name: str = None, data: dict = None
    ):
        """发送日志消息"""
        # 检查日志级别是否满足全局设置
        if self.log_levels.get(level, 0) < self.log_levels.get(
            self.global_log_level, 1
        ):
            return

        # 构造日志消息
        log_data = {
            "message": message,
            "timestamp": time.time(),
            "server_time": datetime.datetime.now().isoformat(),
        }

        # 如果有额外数据，合并到日志数据中
        if data:
            log_data.update(data)

        # 记录到本地日志
        log_method = getattr(logger, level, logger.info)
        log_method(f"[{logger_name or 'server'}] {message}")

        # 返回日志消息结构（用于发送给客户端）
        return {"level": level, "logger": logger_name or "server", "data": log_data}

    # 设置日志级别
+   def set_log_level(self, level: str) -> bool:
+       """设置日志级别"""
+       if level in self.log_levels:
+           self.global_log_level = level
+           logger.info(f"日志级别已设置为: {level}")
+           return True
+       else:
+           logger.error(f"无效的日志级别: {level}")
+           return False

    # 处理日志级别设置请求
+   def handle_logging_set_level(self, params, session_id: str = None):
+       """处理日志级别设置请求"""
+       level = params.get("level")
+       if not level:
+           return {
+               "error": {
+                   "code": -32602,
+                   "message": "Missing required parameter: level",
+               }
+           }

+       success = self.set_log_level(level)
+       if success:
+           return {"success": True, "level": level}
+       else:
+           return {"error": {"code": -32602, "message": f"Invalid log level: {level}"}}

    # 获取所有活跃会话信息
    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 _cleanup_expired_sessions(self, max_age: int = 3600):
        """清理过期会话（默认1小时）"""
        current_time = time.time()
        expired_sessions = []

        for session_id, session_data in self.sessions.items():
            if current_time - session_data["last_activity"] > max_age:
                expired_sessions.append(session_id)

        for session_id in expired_sessions:
            del self.sessions[session_id]
            logger.info(f"清理过期会话: {session_id}")

        return len(expired_sessions)

    # 获取或创建会话
    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)

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

    # 处理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}，开始发送时间更新")
            # 发送一些示例日志消息
            self._send_log_event(
                self.mcp_server.send_log_message(
                    "info",
                    "SSE连接已建立",
                    "sse",
                    {"session_id": session_id, "connection_type": "SSE"},
                )
            )
            # 定时发送服务器时间
            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
+       elif method == "logging/setLevel":
            # 处理日志级别设置请求
+           result = self.mcp_server.handle_logging_set_level(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)

    # 创建ThreadingHTTPServer实例，绑定主机和端口
    server = ThreadingHTTPServer((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()