from fastmcp import FastMCP

# 1. 创建服务器实例，名称为"My First MCP Server"
mcp = FastMCP("My First MCP Server")

# 2. 注册一个工具：让LLM可以调用的打招呼函数
@mcp.tool
def greet(name: str) -> str:
    """向指定名字的人发送问候语"""
    return f"Hello, {name}! 欢迎使用FastMCP"

# 3. 启动服务器
if __name__ == "__main__":
    # 默认使用STDIO传输，适合本地调试和Claude Desktop集成
    mcp.run()

if __name__ == "__main__":
    # HTTP模式，端口8000，支持远程访问
    mcp.run(transport="http", port=8000)

import asyncio
from fastmcp import Client

async def test_server():
    # 连接到本地HTTP服务器
    client = Client("http://localhost:8000/mcp")

    async with client:
        # 1. 列出所有可用工具
        tools = await client.list_tools()
        print("可用工具：")
        for tool in tools:
            print(f"- {tool.name}: {tool.description}")

        # 2. 调用greet工具
        result = await client.call_tool("greet", {"name": "FastMCP用户"})
        print(f"\n工具调用结果：{result}")

if __name__ == "__main__":
    asyncio.run(test_server())

from typing import Annotated
from pydantic import Field
from fastmcp import FastMCP

mcp = FastMCP("数据处理工具集")

# 基础工具：同步函数
@mcp.tool
def calculate_sum(numbers: list[float]) -> float:
    """计算一组数字的总和"""
    return sum(numbers)

# 进阶工具：带参数描述、异步函数
@mcp.tool
async def get_weather(
    city: Annotated[str, Field(description="要查询的城市名称，如深圳、北京")],
    province: Annotated[str, Field(description="省份名称，可选，用于区分重名城市")] = ""
) -> str:
    """获取指定城市的实时天气信息"""
    # 这里可替换为真实的天气API调用
    await asyncio.sleep(0.1) # 模拟IO延迟
    return f"{province}{city} 当前天气：晴，25℃，微风"

# 工具错误处理：主动抛出ToolError，LLM可识别错误信息
from fastmcp.exceptions import ToolError

@mcp.tool
def divide(a: float, b: float) -> float:
    """两个数字的除法运算"""
    if b == 0:
        raise ToolError("除数不能为0")
    return a / b

from fastmcp import FastMCP

mcp = FastMCP("用户数据服务")

# 静态资源：固定URI，返回固定内容
@mcp.resource("config://server/info")
def get_server_info() -> str:
    """服务器基础信息"""
    return "FastMCP 服务器 v1.0，运行于Python 3.12"

# 动态资源：URI模板，支持路径参数，动态生成内容
@mcp.resource("user://{user_id}/profile")
def get_user_profile(user_id: int) -> dict:
    """获取指定用户的个人资料"""
    # 模拟数据库查询
    user_db = {
        1: {"name": "张三", "age": 28, "role": "管理员"},
        2: {"name": "李四", "age": 24, "role": "普通用户"}
    }
    return user_db.get(user_id, {"error": "用户不存在"})

async def read_resource():
    client = Client("http://localhost:8000/mcp")
    async with client:
        # 读取静态资源
        server_info = await client.read_resource("config://server/info")
        print("服务器信息：", server_info)

        # 读取动态资源
        user_profile = await client.read_resource("user://1/profile")
        print("用户资料：", user_profile)

from fastmcp import FastMCP
from fastmcp.prompts import Message

mcp = FastMCP("代码助手提示模板")

# 基础提示：返回字符串，自动转为用户消息
@mcp.prompt
def debug_code(error_log: str) -> str:
    """生成代码调试提示"""
    return f"""
    请分析以下错误日志，给出详细的原因分析和可直接运行的修复代码：
    {error_log}
    要求：
    1. 先说明错误根因
    2. 给出完整的修复代码
    3. 补充注意事项
    """

# 进阶提示：返回多轮消息，支持指定角色
@mcp.prompt
def code_review(language: str, code: str) -> list[Message]:
    """生成代码评审对话模板"""
    return [
        Message(role="user", content=f"请帮我评审以下{language}代码：\n{code}"),
        Message(role="assistant", content="我会从代码规范、性能、安全性、可维护性四个维度进行评审，给出具体的优化建议。"),
    ]

async def get_prompt():
    client = Client("http://localhost:8000/mcp")
    async with client:
        # 获取调试代码的提示模板，传入参数
        prompt = await client.get_prompt(
            "debug_code",
            variables={"error_log": "IndexError: list index out of range"}
        )
        print("渲染后的提示：", prompt.messages)

import os
import logging
import asyncio
from typing import Annotated, List, Dict, Any
from datetime import datetime
from dotenv import load_dotenv

from fastmcp import FastMCP
from fastmcp.exceptions import ToolError
from fastmcp.server.middleware import Middleware, MiddlewareContext
from fastmcp.server.dependencies import get_http_headers
from pydantic import Field

# ==========================================
# 1. 基础配置与环境加载
# ==========================================
# 加载环境变量（本地开发时创建 .env 文件，生产环境通过系统环境变量注入）
load_dotenv()

# ==========================================
# 2. 结构化日志配置
# ==========================================
def setup_logging() -> logging.Logger:
    """配置生产级结构化日志"""
    logger = logging.getLogger("fastmcp_production")
    logger.setLevel(logging.INFO)

    # 避免重复添加处理器
    if logger.handlers:
        return logger

    # 日志格式：时间 | 日志级别 | 模块名 | 消息内容
    formatter = logging.Formatter(
        "%(asctime)s | %(levelname)-8s | %(name)s:%(lineno)d | %(message)s",
        datefmt="%Y-%m-%d %H:%M:%S"
    )

    # 控制台输出
    console_handler = logging.StreamHandler()
    console_handler.setFormatter(formatter)
    logger.addHandler(console_handler)

    # 文件输出（生产环境建议使用日志收集系统，如 ELK/Loki）
    log_file = os.getenv("LOG_FILE", "fastmcp_server.log")
    file_handler = logging.FileHandler(log_file, encoding="utf-8")
    file_handler.setFormatter(formatter)
    logger.addHandler(file_handler)

    return logger

logger = setup_logging()

# ==========================================
# 3. API Key 认证中间件
# ==========================================
class ApiKeyAuthMiddleware(Middleware):
    """API Key 认证中间件，保护所有工具和资源访问"""

    def __init__(self):
        # 从环境变量读取有效 API Key，支持多个 Key（逗号分隔）
        valid_keys_str = os.getenv("VALID_API_KEYS", "")
        self.valid_api_keys = set(key.strip() for key in valid_keys_str.split(",") if key.strip())
        self.auth_disabled = os.getenv("DISABLE_AUTH", "false").lower() == "true"

        if self.auth_disabled:
            logger.warning("⚠️  认证已禁用，仅用于本地开发！")
        elif not self.valid_api_keys:
            logger.error("❌ 未配置 VALID_API_KEYS 环境变量，认证将失败！")

    async def on_request(self, context: MiddlewareContext, call_next):
        """全局请求拦截：仅在 HTTP 模式下校验 API Key"""
        # STDIO 模式（本地 Claude Desktop）跳过认证
        if context.transport == "stdio":
            return await call_next(context)

        # 认证已禁用
        if self.auth_disabled:
            return await call_next(context)

        # HTTP 模式：校验 API Key
        headers = get_http_headers() or {}
        api_key = headers.get("x-api-key")

        if not api_key:
            logger.warning("🚫 未提供 API Key，拒绝访问")
            raise ToolError("未提供 API Key，请在请求头中添加 x-api-key")

        if api_key not in self.valid_api_keys:
            logger.warning(f"🚫 无效 API Key: {api_key[:8]}...")
            raise ToolError("无效的 API Key")

        logger.info(f"✅ API Key 认证通过: {api_key[:8]}...")
        return await call_next(context)

# ==========================================
# 4. 全局错误处理中间件
# ==========================================
class GlobalErrorHandlerMiddleware(Middleware):
    """全局错误捕获与日志记录中间件"""

    async def on_request(self, context: MiddlewareContext, call_next):
        try:
            return await call_next(context)
        except ToolError:
            # 业务错误直接抛出，不记录为系统错误
            raise
        except Exception as e:
            # 系统错误：记录详细日志，返回友好提示
            logger.exception(f"💥 系统错误 | 方法: {context.method} | 错误: {type(e).__name__}")
            raise ToolError(f"服务器内部错误，请联系管理员。错误ID: {datetime.now().strftime('%Y%m%d%H%M%S')}")

# ==========================================
# 5. 创建 FastMCP 服务器实例
# ==========================================
mcp = FastMCP(
    name="生产级数据服务平台",
    dependencies=["requests>=2.32.0", "pandas>=2.2.0", "python-dotenv>=1.0.0"],
    instructions="""
    这是一个生产级数据服务平台，提供以下能力：
    1. 数据计算与统计分析
    2. 外部 API 数据拉取
    3. 安全的文件操作
    4. 系统配置与状态查询

    调用工具前请先确认用户需求，优先使用本服务的工具完成任务。
    """
)

# 注册中间件（顺序很重要：先错误处理，再认证）
mcp.add_middleware(GlobalErrorHandlerMiddleware())
mcp.add_middleware(ApiKeyAuthMiddleware())

# ==========================================
# 6. 生命周期管理（资源初始化与回收）
# ==========================================
@mcp.on_startup
async def startup_event():
    """服务启动时的资源初始化"""
    logger.info("🚀 FastMCP 服务启动中...")

    # 示例：初始化数据库连接、API 客户端等
    # global db_client
    # db_client = await create_database_connection()

    logger.info("✅ 服务初始化完成")

@mcp.on_shutdown
async def shutdown_event():
    """服务关闭时的资源回收"""
    logger.info("🛑 FastMCP 服务关闭中...")

    # 示例：关闭数据库连接、清理临时文件等
    # await db_client.close()

    logger.info("✅ 资源回收完成，服务已停止")

# ==========================================
# 7. 工具集成（多类型示例）
# ==========================================

# ------------------------------
# 工具1：同步数据计算工具
# ------------------------------
@mcp.tool
def calculate_statistics(numbers: List[float]) -> Dict[str, float]:
    """
    计算一组数字的统计指标：总和、平均值、最大值、最小值

    Args:
        numbers: 待计算的数字列表

    Returns:
        包含统计结果的字典
    """
    if not numbers:
        raise ToolError("数字列表不能为空")

    logger.info(f"📊 执行统计计算，数据量: {len(numbers)}")

    return {
        "sum": sum(numbers),
        "mean": sum(numbers) / len(numbers),
        "max": max(numbers),
        "min": min(numbers),
        "count": len(numbers)
    }

# ------------------------------
# 工具2：异步 API 调用工具
# ------------------------------
@mcp.tool
async def fetch_external_api(
    url: Annotated[str, Field(description="要请求的外部 API URL")],
    timeout: Annotated[int, Field(description="请求超时时间（秒）", ge=1, le=60)] = 10
) -> Dict[str, Any]:
    """
    安全地调用外部 HTTP API（仅支持 GET 请求）

    Args:
        url: 外部 API 的完整 URL
        timeout: 请求超时时间，范围 1-60 秒

    Returns:
        API 返回的 JSON 数据
    """
    import requests

    # 安全限制：仅允许 HTTPS 请求
    if not url.startswith("https://"):
        raise ToolError("仅支持 HTTPS 协议的 API 请求")

    logger.info(f"🌐 调用外部 API: {url}")

    try:
        response = requests.get(url, timeout=timeout)
        response.raise_for_status()  # 抛出 HTTP 错误
        return response.json()
    except requests.exceptions.Timeout:
        raise ToolError(f"API 请求超时（{timeout}秒）")
    except requests.exceptions.HTTPError as e:
        raise ToolError(f"API 请求失败，状态码: {e.response.status_code}")
    except Exception as e:
        logger.exception(f"API 调用异常: {url}")
        raise ToolError(f"API 调用失败: {str(e)}")

# ------------------------------
# 工具3：安全文件操作工具
# ------------------------------
@mcp.tool
def read_safe_file(
    file_path: Annotated[str, Field(description="要读取的文件路径（仅允许在安全目录内）")]
) -> str:
    """
    安全读取文件内容（仅允许访问安全目录内的文件）

    Args:
        file_path: 要读取的文件路径

    Returns:
        文件内容字符串
    """
    # 安全目录配置
    safe_dir = os.getenv("SAFE_FILE_DIR", os.path.expanduser("~/fastmcp_safe"))
    os.makedirs(safe_dir, exist_ok=True)

    # 构建绝对路径并检查是否在安全目录内
    absolute_path = os.path.abspath(os.path.join(safe_dir, file_path))
    if not absolute_path.startswith(os.path.abspath(safe_dir)):
        raise ToolError("非法的文件路径，仅允许访问安全目录内的文件")

    if not os.path.exists(absolute_path):
        raise ToolError(f"文件不存在: {file_path}")

    logger.info(f"📄 读取文件: {absolute_path}")

    with open(absolute_path, "r", encoding="utf-8") as f:
        return f.read()

# ==========================================
# 8. 资源定义（只读数据注入）
# ==========================================

# ------------------------------
# 资源1：静态服务信息
# ------------------------------
@mcp.resource("service://info")
def get_service_info() -> str:
    """服务基础信息资源"""
    return f"""
    服务名称: 生产级数据服务平台
    版本: 1.0.0
    启动时间: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}
    运行环境: {os.getenv('ENVIRONMENT', 'development')}
    """

# ------------------------------
# 资源2：动态用户配置
# ------------------------------
@mcp.resource("config://user/{user_id}")
def get_user_config(user_id: str) -> Dict[str, Any]:
    """
    获取指定用户的配置信息（动态资源）

    Args:
        user_id: 用户ID

    Returns:
        用户配置字典
    """
    # 模拟数据库查询
    user_configs = {
        "admin": {
            "role": "管理员",
            "permissions": ["read", "write", "delete"],
            "theme": "dark"
        },
        "user": {
            "role": "普通用户",
            "permissions": ["read"],
            "theme": "light"
        }
    }

    config = user_configs.get(user_id)
    if not config:
        raise ToolError(f"用户配置不存在: {user_id}")

    logger.info(f"⚙️  读取用户配置: {user_id}")
    return config

# ==========================================
# 9. 服务启动入口
# ==========================================
if __name__ == "__main__":
    # 传输模式配置：通过环境变量切换，默认 STDIO（本地 Claude Desktop）
    transport = os.getenv("TRANSPORT", "stdio")
    port = int(os.getenv("PORT", 8000))

    logger.info(f"🚀 启动服务 | 传输模式: {transport} | 端口: {port}")

    if transport == "stdio":
        # STDIO 模式：本地 Claude Desktop 集成
        mcp.run()
    elif transport == "http":
        # HTTP 模式：生产级远程部署
        import uvicorn
        app = mcp.http_app()
        uvicorn.run(
            app,
            host="0.0.0.0",
            port=port,
            timeout_keep_alive=300,
            log_level="info"
        )
    else:
        logger.error(f"❌ 不支持的传输模式: {transport}")
        raise ValueError(f"无效的传输模式: {transport}，仅支持 stdio 或 http")

import asyncio
from fastmcp import Client

async def test_production_server():
    # 连接到本地 HTTP 服务器
    client = Client(
        "http://localhost:8000/mcp",
        headers={"x-api-key": "your_secure_api_key_123"}
    )

    async with client:
        # 1. 列出所有工具
        tools = await client.list_tools()
        print("可用工具：")
        for tool in tools:
            print(f"- {tool.name}: {tool.description}")

        # 2. 调用统计计算工具
        stats = await client.call_tool(
            "calculate_statistics",
            {"numbers": [1.2, 3.4, 5.6, 7.8]}
        )
        print("\n统计结果：", stats)

        # 3. 读取服务信息资源
        service_info = await client.read_resource("service://info")
        print("\n服务信息：", service_info)

if __name__ == "__main__":
    asyncio.run(test_production_server())

from fastmcp.server.middleware import Middleware, MiddlewareContext
from fastmcp.exceptions import ToolError
# 按需导入其他依赖：如限流用的asyncio、日志用的logging、时间统计用的datetime等

# 自定义中间件示例：请求耗时统计
class CustomTimingMiddleware(Middleware):
    """自定义请求耗时统计中间件"""

    async def on_request(self, context: MiddlewareContext, call_next):
        """所有请求都会经过的预处理+后处理逻辑"""
        # ========== 预处理阶段：请求进入业务逻辑前执行 ==========
        from datetime import datetime
        start_time = datetime.now()
        request_method = context.method
        session_id = context.session_id

        # ========== 执行后续链路：必须调用，否则请求终止 ==========
        response = await call_next(context)

        # ========== 后处理阶段：业务逻辑执行完，响应返回前执行 ==========
        end_time = datetime.now()
        cost_ms = (end_time - start_time).total_seconds() * 1000
        print(f"请求[{request_method}] 会话[{session_id}] 耗时: {cost_ms:.2f}ms")

        # 必须返回响应结果（可修改后返回）
        return response

# 5. 创建 FastMCP 服务器实例（生产模板原有代码）
mcp = FastMCP(
    name="生产级数据服务平台",
    dependencies=["requests>=2.32.0", "pandas>=2.2.0", "python-dotenv>=1.0.0"],
    instructions="..."
)

# ========== 注册自定义中间件（核心步骤） ==========
# 注意：注册顺序决定执行顺序，错误处理中间件必须放在最前面
mcp.add_middleware(GlobalErrorHandlerMiddleware())  # 原有全局错误处理
mcp.add_middleware(CustomTimingMiddleware())         # 新增的自定义耗时统计中间件
mcp.add_middleware(ApiKeyAuthMiddleware())           # 原有认证中间件
# 继续添加其他自定义中间件...

class OperationAuditMiddleware(Middleware):
    """全链路操作审计日志中间件"""

    def __init__(self):
        self.logger = logging.getLogger("fastmcp_audit")

    async def on_call_tool(self, context: MiddlewareContext, call_next):
        """审计工具调用操作"""
        from datetime import datetime
        start_time = datetime.now()
        tool_name = context.message.name
        tool_params = context.message.arguments
        session_id = context.session_id
        transport = context.transport

        self.logger.info(
            f"【工具调用开始】会话ID:{session_id} | 传输模式:{transport} | 工具名:{tool_name} | 入参:{tool_params}"
        )

        try:
            # 执行工具调用
            result = await call_next(context)
            cost_ms = (datetime.now() - start_time).total_seconds() * 1000

            # 记录成功日志
            self.logger.info(
                f"【工具调用成功】会话ID:{session_id} | 工具名:{tool_name} | 耗时:{cost_ms:.2f}ms"
            )
            return result

        except ToolError as e:
            # 记录业务异常
            self.logger.warning(
                f"【工具调用业务异常】会话ID:{session_id} | 工具名:{tool_name} | 错误:{str(e)}"
            )
            raise
        except Exception as e:
            # 记录系统异常
            self.logger.exception(
                f"【工具调用系统异常】会话ID:{session_id} | 工具名:{tool_name} | 错误:{type(e).__name__}"
            )
            raise

    async def on_read_resource(self, context: MiddlewareContext, call_next):
        """审计资源读取操作"""
        resource_uri = context.message.uri
        session_id = context.session_id
        self.logger.info(f"【资源读取】会话ID:{session_id} | URI:{resource_uri}")
        return await call_next(context)

class RateLimitMiddleware(Middleware):
    """工具调用限流中间件：单会话每秒最多调用5次工具"""

    def __init__(self, max_calls: int = 5, time_window: int = 1):
        from collections import defaultdict
        self.max_calls = max_calls  # 时间窗口内最大调用次数
        self.time_window = time_window  # 时间窗口（秒）
        self.session_call_records = defaultdict(list)  # 会话调用记录
        self.logger = logging.getLogger("fastmcp_rate_limit")

    async def on_call_tool(self, context: MiddlewareContext, call_next):
        import time
        session_id = context.session_id
        tool_name = context.message.name
        now = time.time()

        # 清理过期的调用记录
        self.session_call_records[session_id] = [
            call_time for call_time in self.session_call_records[session_id]
            if now - call_time < self.time_window
        ]

        # 检查是否超出限流阈值
        if len(self.session_call_records[session_id]) >= self.max_calls:
            self.logger.warning(
                f"【限流触发】会话ID:{session_id} | 工具名:{tool_name} | 超出每秒{self.max_calls}次的调用限制"
            )
            raise ToolError(f"工具调用频率过高，请{self.time_window}秒后重试")

        # 记录本次调用
        self.session_call_records[session_id].append(now)
        # 执行后续链路
        return await call_next(context)

class FineGrainedAuthMiddleware(Middleware):
    """细粒度权限控制中间件：按角色限制工具/资源访问"""

    def __init__(self):
        # 角色权限配置：角色名 → 允许调用的工具列表、允许访问的资源前缀
        self.role_permission_map = {
            "admin": {
                "allowed_tools": ["*"],  # * 代表全部允许
                "allowed_resource_prefixes": ["*"]
            },
            "user": {
                "allowed_tools": ["calculate_statistics", "read_safe_file"],
                "allowed_resource_prefixes": ["service://", "config://user/user"]
            },
            "guest": {
                "allowed_tools": ["calculate_statistics"],
                "allowed_resource_prefixes": ["service://info"]
            }
        }
        # API Key 与角色的映射（生产环境从数据库/配置中心读取）
        self.api_key_role_map = {
            os.getenv("ADMIN_API_KEY", ""): "admin",
            os.getenv("USER_API_KEY", ""): "user",
            os.getenv("GUEST_API_KEY", ""): "guest"
        }
        self.logger = logging.getLogger("fastmcp_auth")

    def _get_role_by_api_key(self) -> str:
        """从请求头获取API Key，返回对应角色"""
        headers = get_http_headers() or {}
        api_key = headers.get("x-api-key", "")
        return self.api_key_role_map.get(api_key, "guest")

    async def on_call_tool(self, context: MiddlewareContext, call_next):
        """工具调用权限校验"""
        # STDIO 本地模式跳过权限校验
        if context.transport == "stdio":
            return await call_next(context)

        tool_name = context.message.name
        role = self._get_role_by_api_key()
        permission = self.role_permission_map[role]

        # 校验工具权限
        allowed_tools = permission["allowed_tools"]
        if "*" not in allowed_tools and tool_name not in allowed_tools:
            self.logger.warning(f"【权限拒绝】角色:{role} 无权限调用工具:{tool_name}")
            raise ToolError(f"您的角色[{role}]无权限调用工具[{tool_name}]")

        self.logger.info(f"【权限通过】角色:{role} 调用工具:{tool_name}")
        return await call_next(context)

    async def on_read_resource(self, context: MiddlewareContext, call_next):
        """资源读取权限校验"""
        if context.transport == "stdio":
            return await call_next(context)

        resource_uri = context.message.uri
        role = self._get_role_by_api_key()
        permission = self.role_permission_map[role]

        # 校验资源权限
        allowed_prefixes = permission["allowed_resource_prefixes"]
        if "*" not in allowed_prefixes:
            has_permission = any(resource_uri.startswith(prefix) for prefix in allowed_prefixes)
            if not has_permission:
                self.logger.warning(f"【权限拒绝】角色:{role} 无权限访问资源:{resource_uri}")
                raise ToolError(f"您的角色[{role}]无权限访问资源[{resource_uri}]")

        return await call_next(context)

    async def on_list_tools(self, context: MiddlewareContext, call_next):
        """按角色过滤可见工具列表，无权限的工具不展示给LLM"""
        if context.transport == "stdio":
            return await call_next(context)

        role = self._get_role_by_api_key()
        permission = self.role_permission_map[role]
        allowed_tools = permission["allowed_tools"]

        # 获取完整工具列表
        all_tools = await call_next(context)

        # 过滤工具列表
        if "*" in allowed_tools:
            return all_tools
        return [tool for tool in all_tools if tool.name in allowed_tools]

class DataDesensitizationMiddleware(Middleware):
    """响应数据脱敏中间件：自动脱敏手机号、身份证号等敏感信息"""

    def _desensitize_content(self, content: str | dict | list) -> any:
        """递归脱敏敏感数据"""
        import re

        if isinstance(content, str):
            # 手机号脱敏
            content = re.sub(r'1[3-9]\d{9}', lambda m: m.group()[:3] + '****' + m.group()[-4:], content)
            # 身份证号脱敏
            content = re.sub(r'\d{17}[\dXx]', lambda m: m.group()[:6] + '********' + m.group()[-4:], content)
            return content

        elif isinstance(content, dict):
            return {k: self._desensitize_content(v) for k, v in content.items()}

        elif isinstance(content, list):
            return [self._desensitize_content(item) for item in content]

        else:
            return content

    async def on_read_resource(self, context: MiddlewareContext, call_next):
        """资源读取响应脱敏"""
        # 执行资源读取
        resource_content = await call_next(context)
        # 脱敏后返回
        return self._desensitize_content(resource_content)

    async def on_call_tool(self, context: MiddlewareContext, call_next):
        """工具调用响应脱敏"""
        result = await call_next(context)
        return self._desensitize_content(result)

# my_server.py
if __name__ == "__main__":
    import uvicorn
    # 生成ASGI应用
    app = mcp.http_app()
    uvicorn.run(app, host="0.0.0.0", port=8000, timeout_keep_alive=300)