1. 连接服务器 #

1.1. pagination.py #

pagination.py

from client import MCPClient

def pagination():
    """简单分页演示"""
    print(" MCP分页功能")

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

    try:
        # 初始化连接
        if not client.initialize():
            print(" 连接失败")
            return
        print(" 连接成功")
    except Exception as e:
        print(f" 演示失败: {e}")
    finally:
        client.reset_session()


if __name__ == "__main__":
    pagination()

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",
                "MCP-Protocol-Version": self.protocol_version,
            }
        )

    # 重置会话方法
    def reset_session(self):
        """重置会话"""
        # 保存旧的会话ID
        old_session_id = self.session_id
        # 清空当前会话ID
        self.session_id = None
        # 记录会话重置信息
        logger.info(f"会话已重置，原会话ID: {old_session_id}")

    # 获取当前会话信息方法
    def get_session_info(self):
        """获取当前会话信息"""
        # 返回会话相关信息字典
        return {
            "session_id": self.session_id,
            "server_url": self.server_url,
            "protocol_version": self.protocol_version,
        }

    # 获取可用工具列表方法
+   def list_tools(self, cursor=None, page_size=10):
        """获取可用工具列表"""
        # 构造请求消息
        request = {
            "jsonrpc": "2.0",
            "id": self._generate_id(),
            "method": "tools/list",
+           "params": {"pageSize": page_size},
        }
        # 如果有游标，添加到参数中
+       if cursor:
+           request["params"]["cursor"] = cursor
        # 发送请求并获取响应
        response = self._send_message(request)
        # 如果响应存在且包含result字段
        if response and "result" in response:
            # 获取工具列表和分页信息
+           result = response["result"]
            # 获取工具列表和分页信息
+           tools = result.get("tools", [])
            # 获取下一页游标
+           next_cursor = result.get("nextCursor")
            # 获取当前页码
+           page = result.get("page", 1)
            # 获取总页数
+           total = result.get("total", 0)
            # 获取是否有下一页
+           has_next = result.get("hasNext", False)

+           logger.info(
+               f"工具列表第{page}页: {len(tools)}/{total} 个工具, 有下一页: {has_next}"
+           )
+           return {
+               "tools": tools,
+               "nextCursor": next_cursor,
+               "page": page,
+               "total": total,
+               "hasNext": has_next,
+           }
        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传输 ===")

    client = MCPClient("http://localhost:8000")

    try:
        if not client.initialize():
            print("初始化失败")
            return
        session_info = client.get_session_info()
        print(f" 会话信息: {session_info}")
        tools = client.list_tools()
        print(f"可用工具: {[tool['name'] for tool in tools]}")

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

        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()

2.2. pagination.py #

pagination.py

from client import MCPClient


def pagination():
    """简单分页演示"""
    print(" MCP分页功能")

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

    try:
        # 初始化连接
        if not client.initialize():
            print(" 连接失败")
            return

        print(" 连接成功")
        # 演示工具分页
+       print("\n🔧 工具分页演示:")
+       result = client.list_tools(page_size=5)
+       print(f"第{result['page']}页: {len(result['tools'])}/{result['total']} 个工具")

+       if result["hasNext"]:
+           print("获取下一页...")
+           next_result = client.list_tools(cursor=result["nextCursor"], page_size=5)
+           print(
+               f"第{next_result['page']}页: {len(next_result['tools'])}/{next_result['total']} 个工具"
+           )
    except Exception as e:
        print(f" 演示失败: {e}")
    finally:
        client.reset_session()


if __name__ == "__main__":
    pagination()

2.3. server.py #

server.py

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

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

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

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

# 从tools模块导入tools对象
+from tools import tools, generate_tools_data

# 导入datetime模块，用于处理日期和时间
# 导入base64模块，用于编码游标
+import base64

# 从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": {}}
        # 会话管理字典
        self.sessions = {}
        # 初始化分页数据
+       self._init_pagination_data()

+   def _init_pagination_data(self):
+       """初始化分页数据"""
+       self.tools_data = generate_tools_data()

+       logger.info(f"初始化分页数据: {len(self.tools_data)} 个工具")

    # 获取所有活跃会话信息
    def get_all_sessions(self):
        """获取所有活跃会话信息"""
        return self.sessions.copy()

    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 handle_tools_list(self, params=None, session_id: str = None):
        """处理工具列表请求"""
        # 如果有会话ID，则获取或创建会话
        if session_id:
            self._get_or_create_session(session_id)
        # 获取分页参数
+       cursor = params.get("cursor") if params else None
        # 获取分页大小
+       page_size = params.get("pageSize", 10) if params else 10
        # 执行分页
+       paginated_result = self._paginate_data(self.tools_data, cursor, page_size)
        # 返回分页结果
+       return {
+           "tools": paginated_result["data"],
+           "nextCursor": paginated_result["nextCursor"],
+           "page": paginated_result["page"],
+           "pageSize": paginated_result["pageSize"],
+           "total": paginated_result["total"],
+           "hasNext": paginated_result["hasNext"],
+       }

    # 定义解析分页游标的方法，接收一个字符串类型的游标参数，返回字典类型
+   def _parse_cursor(self, cursor: str) -> dict:
        # 方法说明：解析分页游标
+       """解析分页游标"""
+       try:
            # 尝试对游标进行base64解码，然后将解码后的字节流转为字符串，再用json解析为字典
+           cursor_data = json.loads(base64.b64decode(cursor).decode())
            # 返回解析后的游标数据字典
+           return cursor_data
+       except Exception as e:
            # 如果解析过程中发生异常，记录错误日志
+           logger.error(f"游标解析错误: {e}")
            # 返回默认的游标信息，页码为1，每页10条，数据类型为tools
+           return {"page": 1, "page_size": 10, "data_type": "tools"}

    # 定义生成分页游标的方法，接收页码、每页数量和数据类型，返回字符串类型的游标
+   def _generate_cursor(self, page: int, page_size: int, data_type: str) -> str:
        # 方法说明：生成分页游标
+       """生成分页游标"""
        # 构造游标数据字典，包含当前页码、每页数量、数据类型和时间戳
+       cursor_data = {
+           "page": page,
+           "page_size": page_size,
+           "data_type": data_type,
+           "timestamp": time.time(),
+       }
        # 将游标数据字典转为JSON字符串后编码为字节流，再用base64编码为字符串返回
+       return base64.b64encode(json.dumps(cursor_data).encode()).decode()

    # 定义分页处理数据的方法
+   def _paginate_data(
+       self, data: list, cursor: str = None, page_size: int = 10
+   ) -> dict:
        # 方法说明：分页处理数据
+       """分页处理数据"""
        # 如果没有游标，说明是第一页
+       if not cursor:
            # 设置当前页为1
+           page = 1
+           data_type = "tools"
+       else:
            # 解析游标，获取页码等信息
+           cursor_data = self._parse_cursor(cursor)
            # 获取当前页码，默认为1
+           page = cursor_data.get("page", 1)
            # 获取每页大小，优先使用游标中的page_size
+           page_size = cursor_data.get("page_size", page_size)
            # 获取数据类型，默认为"tools"
+           data_type = cursor_data.get("data_type", "tools")

        # 计算当前页的起始索引
+       start_index = (page - 1) * page_size
        # 计算当前页的结束索引
+       end_index = start_index + page_size
        # 获取当前页的数据
+       page_data = data[start_index:end_index]

        # 判断是否还有下一页
+       has_next = end_index < len(data)
        # 初始化下一页游标为None
+       next_cursor = None
        # 如果有下一页，则生成下一页的游标
+       if has_next:
+           next_cursor = self._generate_cursor(page + 1, page_size, data_type)

        # 记录分页日志信息
+       logger.info(
+           f"分页数据: 第{page}页, 大小{page_size}, 数据量{len(page_data)}, 总计{len(data)}, 有下一页: {has_next}"
+       )

        # 返回分页结果的字典
+       return {
+           "data": page_data,
+           "nextCursor": next_cursor,
+           "page": page,
+           "pageSize": page_size,
+           "total": len(data),
+           "hasNext": has_next,
+       }

    # 处理初始化请求（旧方法，已被覆盖）
    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)

    # 处理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
            accept_header = self.headers.get("Accept", "")
            if "application/json" not in accept_header:
                self.send_error(400, "Missing required Accept header")
                return
            # 检查MCP-Protocol-Version头是否匹配
            protocol_version = self.headers.get("MCP-Protocol-Version")
            if (
                protocol_version
                and protocol_version != self.mcp_server.protocol_version
            ):
                self.send_error(
                    400, f"Unsupported protocol version: {protocol_version}"
                )
                return
            # 获取会话ID
            session_id = self.headers.get("Mcp-Session-Id")
            # 读取请求体长度
            content_length = int(self.headers.get("Content-Length", 0))
            # 读取请求体内容
            body = self.rfile.read(content_length)
            try:
                # 尝试解析JSON消息
                message = json.loads(body.decode("utf-8"))
            except json.JSONDecodeError:
                # JSON解析失败
                self.send_error(400, "Invalid JSON")
                return
            # 处理MCP消息
            response = self._handle_mcp_message(message, session_id)
            # 发送响应
            if response:
                # 发送200响应码
                self.send_response(200)
                # 设置响应头Content-Type为application/json
                self.send_header("Content-Type", "application/json")
                # 设置协议版本头
                self.send_header(
                    "MCP-Protocol-Version", self.mcp_server.protocol_version
                )
                # 如果响应中包含会话ID，则添加到响应头
                if "result" in response and "sessionId" in response["result"]:
                    self.send_header("Mcp-Session-Id", response["result"]["sessionId"])
                # 结束响应头
                self.end_headers()
                # 写入JSON响应体
                self.wfile.write(json.dumps(response).encode("utf-8"))
            else:
                # 没有响应内容，返回202 Accepted
                self.send_response(202)  # Accepted
                self.end_headers()
        except Exception as e:
            # 记录错误日志
            logger.error(f"POST处理错误: {e}")
            # 返回500内部服务器错误
            self.send_error(500, "Internal Server Error")


# 定义运行HTTP服务器的函数
def run_http_server(mcp_server: MCPServer, host: str = "localhost", port: int = 8000):
    """运行HTTP服务器"""

    # 定义处理器工厂函数，用于传递mcp_server实例
    def handler_factory(*args, **kwargs):
        return MCPHTTPHandler(*args, mcp_server=mcp_server, **kwargs)

    # 创建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.4. tools.py #

tools.py

# MCP工具定义


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


# 生成大量工具数据用于演示分页
+def generate_tools_data():
+   """生成大量工具数据用于演示分页"""
+   tools_data = []
+   for i in range(100):  # 生成100个工具
+       tools_data.append(
+           {
+               "name": f"tool_{i:03d}",
+               "description": f"这是第{i+1}个工具，用于演示分页功能",
+               "inputSchema": {
+                   "type": "object",
+                   "properties": {
+                       "param": {
+                           "type": "string",
+                           "description": f"工具{i+1}的参数",
+                       }
+                   },
+                   "required": ["param"],
+               },
+           }
+       )
+   return tools_data