08c - 自定义 MCP Server 开发

# 创建项目目录
mkdir my-mcp-server && cd my-mcp-server
 
# 初始化 Node.js 项目
npm init -y
 
# 安装 MCP SDK 和 Zod（Schema 验证）
npm install @modelcontextprotocol/sdk zod
 
# 安装开发依赖
npm install -D typescript @types/node

{
  "compilerOptions": {
    "target": "ES2022",
    "module": "Node16",
    "moduleResolution": "Node16",
    "outDir": "./dist",
    "rootDir": "./src",
    "strict": true,
    "esModuleInterop": true,
    "skipLibCheck": true,
    "declaration": true
  },
  "include": ["src/**/*"]
}

{
  "name": "my-mcp-server",
  "version": "1.0.0",
  "type": "module",
  "bin": {
    "my-mcp-server": "./dist/index.js"
  },
  "scripts": {
    "build": "tsc",
    "start": "node dist/index.js",
    "dev": "tsc --watch"
  }
}

#!/usr/bin/env node
 
import { McpServer } from "@modelcontextprotocol/sdk/server/mcp.js";
import { StdioServerTransport } from "@modelcontextprotocol/sdk/server/stdio.js";
 
// 创建 MCP Server 实例
const server = new McpServer({
  name: "my-mcp-server",       // Server 名称（用于能力协商）
  version: "1.0.0",            // Server 版本
  description: "一个示例 MCP Server，展示工具、资源和 Prompt 模板"
});
 
// 使用 stdio 传输启动 Server
async function main() {
  const transport = new StdioServerTransport();
  await server.connect(transport);
  console.error("MCP Server 已启动（stdio 模式）");  // 日志输出到 stderr
}
 
main().catch(console.error);

import { z } from "zod";
 
// ========== 工具 1：字符串处理工具 ==========
server.tool(
  "text_stats",                              // 工具名称
  "分析文本的统计信息（字数、字符数、行数）",      // 工具描述（AI 模型据此选择工具）
  {
    // 使用 Zod 定义输入参数 Schema
    text: z.string().describe("要分析的文本内容"),
    include_whitespace: z.boolean()
      .optional()
      .default(false)
      .describe("是否在字符计数中包含空白字符")
  },
  async ({ text, include_whitespace }) => {
    // 工具执行逻辑
    const lines = text.split("\n").length;
    const words = text.split(/\s+/).filter(w => w.length > 0).length;
    const chars = include_whitespace ? text.length : text.replace(/\s/g, "").length;
 
    return {
      content: [
        {
          type: "text" as const,
          text: JSON.stringify({ lines, words, chars }, null, 2)
        }
      ]
    };
  }
);
 
// ========== 工具 2：带外部 API 调用的工具 ==========
server.tool(
  "get_weather",
  "获取指定城市的当前天气信息",
  {
    city: z.string().describe("城市名称（如：北京、上海、Tokyo）"),
    unit: z.enum(["celsius", "fahrenheit"])
      .optional()
      .default("celsius")
      .describe("温度单位")
  },
  async ({ city, unit }) => {
    try {
      // 实际项目中调用真实天气 API
      // 这里用模拟数据演示
      const weatherData = {
        city,
        temperature: unit === "celsius" ? 25 : 77,
        unit,
        condition: "晴",
        humidity: 45,
        wind_speed: "12 km/h"
      };
 
      return {
        content: [
          {
            type: "text" as const,
            text: JSON.stringify(weatherData, null, 2)
          }
        ]
      };
    } catch (error) {
      return {
        content: [
          {
            type: "text" as const,
            text: `获取天气信息失败: ${error instanceof Error ? error.message : "未知错误"}`
          }
        ],
        isError: true  // 标记为错误结果
      };
    }
  }
);
 
// ========== 工具 3：带结构化输出的工具 ==========
server.tool(
  "generate_uuid",
  "生成指定数量的 UUID v4",
  {
    count: z.number()
      .int()
      .min(1)
      .max(100)
      .default(1)
      .describe("要生成的 UUID 数量（1-100）")
  },
  async ({ count }) => {
    const uuids = Array.from({ length: count }, () => crypto.randomUUID());
 
    return {
      content: [
        {
          type: "text" as const,
          text: JSON.stringify({ uuids, count: uuids.length }, null, 2)
        }
      ]
    };
  }
);

// ========== 静态资源：固定 URI ==========
server.resource(
  "server-config",                           // 资源名称
  "config://server",                         // 资源 URI
  {
    description: "当前 Server 的配置信息",     // 资源描述
    mimeType: "application/json"             // MIME 类型
  },
  async (uri) => {
    const config = {
      name: "my-mcp-server",
      version: "1.0.0",
      uptime: process.uptime(),
      nodeVersion: process.version,
      platform: process.platform,
      tools: ["text_stats", "get_weather", "generate_uuid"]
    };
 
    return {
      contents: [
        {
          uri: uri.href,
          mimeType: "application/json",
          text: JSON.stringify(config, null, 2)
        }
      ]
    };
  }
);
 
// ========== 动态资源模板：URI 中包含参数 ==========
import { ResourceTemplate } from "@modelcontextprotocol/sdk/server/mcp.js";
 
server.resource(
  "user-profile",
  new ResourceTemplate("users://{userId}/profile", {
    list: async () => {
      // 返回可用的资源实例列表（供 AI 浏览）
      const userIds = ["user-001", "user-002", "user-003"];
      return {
        resources: userIds.map(id => ({
          uri: `users://${id}/profile`,
          name: `用户 ${id} 的资料`,
          description: `查看用户 ${id} 的详细资料`
        }))
      };
    }
  }),
  {
    description: "用户资料信息",
    mimeType: "application/json"
  },
  async (uri, { userId }) => {
    // 实际项目中从数据库查询
    const profiles: Record<string, object> = {
      "user-001": { name: "张三", role: "开发者", team: "后端" },
      "user-002": { name: "李四", role: "设计师", team: "产品" },
      "user-003": { name: "王五", role: "运维", team: "基础设施" }
    };
 
    const profile = profiles[userId as string];
    if (!profile) {
      throw new Error(`用户 ${userId} 不存在`);
    }
 
    return {
      contents: [
        {
          uri: uri.href,
          mimeType: "application/json",
          text: JSON.stringify(profile, null, 2)
        }
      ]
    };
  }
);

// ========== Prompt 模板 1：代码审查 ==========
server.prompt(
  "code_review",                             // 模板名称
  "对代码进行专业审查，提供改进建议",            // 模板描述
  {
    // 模板参数
    code: z.string().describe("要审查的代码"),
    language: z.string()
      .optional()
      .default("auto")
      .describe("编程语言（auto 表示自动检测）"),
    focus: z.enum(["security", "performance", "readability", "all"])
      .optional()
      .default("all")
      .describe("审查重点")
  },
  async ({ code, language, focus }) => {
    const focusMap: Record<string, string> = {
      security: "安全漏洞（SQL 注入、XSS、认证绕过等）",
      performance: "性能问题（时间复杂度、内存泄漏、N+1 查询等）",
      readability: "可读性（命名规范、代码结构、注释质量等）",
      all: "安全性、性能和可读性"
    };
 
    return {
      messages: [
        {
          role: "user" as const,
          content: {
            type: "text" as const,
            text: [
              `请对以下${language !== "auto" ? ` ${language} ` : ""}代码进行专业审查。`,
              ``,
              `## 审查重点`,
              `${focusMap[focus ?? "all"]}`,
              ``,
              `## 代码`,
              "```" + (language !== "auto" ? language : ""),
              code,
              "```",
              ``,
              `## 请按以下格式输出`,
              `1. **总体评价**：代码质量评分（1-10）和一句话总结`,
              `2. **问题列表**：按严重程度排序，每个问题包含位置、描述、修复建议`,
              `3. **改进建议**：3-5 条具体的改进建议`,
              `4. **优点**：代码中做得好的地方`
            ].join("\n")
          }
        }
      ]
    };
  }
);
 
// ========== Prompt 模板 2：SQL 生成 ==========
server.prompt(
  "generate_sql",
  "根据自然语言描述生成 SQL 查询",
  {
    description: z.string().describe("用自然语言描述你想查询的数据"),
    dialect: z.enum(["postgresql", "mysql", "sqlite"])
      .optional()
      .default("postgresql")
      .describe("SQL 方言"),
    tables: z.string()
      .optional()
      .describe("相关表结构描述（可选，提供后生成更准确）")
  },
  async ({ description, dialect, tables }) => {
    const messages: Array<{
      role: "user" | "assistant";
      content: { type: "text"; text: string };
    }> = [];
 
    // 系统上下文（通过 assistant 消息提供）
    messages.push({
      role: "assistant" as const,
      content: {
        type: "text" as const,
        text: `我是一个 SQL 专家，擅长 ${dialect?.toUpperCase()} 方言。我会根据你的描述生成安全、高效的 SQL 查询。`
      }
    });
 
    // 用户请求
    let userText = `请根据以下描述生成 ${dialect?.toUpperCase()} 查询：\n\n${description}`;
    if (tables) {
      userText += `\n\n相关表结构：\n${tables}`;
    }
    userText += "\n\n请提供：\n1. SQL 查询语句\n2. 查询说明\n3. 性能优化建议（如需要）";
 
    messages.push({
      role: "user" as const,
      content: { type: "text" as const, text: userText }
    });
 
    return { messages };
  }
);

my-mcp-server/
├── package.json
├── tsconfig.json
├── src/
│   └── index.ts          # 入口文件（包含 Server、Tools、Resources、Prompts）
└── dist/                  # 编译输出
    └── index.js

# 编译 TypeScript
npm run build
 
# 直接运行测试
node dist/index.js

# 创建项目
uv init my-mcp-server
cd my-mcp-server
 
# 添加 MCP SDK 依赖
uv add "mcp[cli]"

# 创建虚拟环境
python -m venv venv
source venv/bin/activate  # Windows: venv\Scripts\activate
 
# 安装 MCP SDK
pip install "mcp[cli]"

my-mcp-server/
├── pyproject.toml
├── server.py              # 主入口文件
└── README.md

#!/usr/bin/env python3
"""一个示例 MCP Server，展示工具、资源和 Prompt 模板。"""
 
from mcp.server.fastmcp import FastMCP
 
# 创建 FastMCP Server 实例
mcp = FastMCP(
    name="my-mcp-server",
    version="1.0.0",
    description="一个示例 MCP Server，展示工具、资源和 Prompt 模板"
)
 
if __name__ == "__main__":
    mcp.run()  # 默认使用 stdio 传输

import json
import uuid
from datetime import datetime, timezone
 
 
@mcp.tool()
def text_stats(text: str, include_whitespace: bool = False) -> str:
    """分析文本的统计信息（字数、字符数、行数）。
 
    Args:
        text: 要分析的文本内容
        include_whitespace: 是否在字符计数中包含空白字符
    """
    lines = len(text.split("\n"))
    words = len(text.split())
    chars = len(text) if include_whitespace else len(text.replace(" ", "").replace("\n", "").replace("\t", ""))
 
    return json.dumps({"lines": lines, "words": words, "chars": chars}, indent=2)
 
 
@mcp.tool()
def get_weather(city: str, unit: str = "celsius") -> str:
    """获取指定城市的当前天气信息。
 
    Args:
        city: 城市名称（如：北京、上海、Tokyo）
        unit: 温度单位，可选 celsius 或 fahrenheit
    """
    # 实际项目中调用真实天气 API
    weather_data = {
        "city": city,
        "temperature": 25 if unit == "celsius" else 77,
        "unit": unit,
        "condition": "晴",
        "humidity": 45,
        "wind_speed": "12 km/h",
        "queried_at": datetime.now(timezone.utc).isoformat()
    }
    return json.dumps(weather_data, indent=2, ensure_ascii=False)
 
 
@mcp.tool()
def generate_uuid(count: int = 1) -> str:
    """生成指定数量的 UUID v4。
 
    Args:
        count: 要生成的 UUID 数量（1-100）
    """
    if not 1 <= count <= 100:
        raise ValueError("count 必须在 1 到 100 之间")
 
    uuids = [str(uuid.uuid4()) for _ in range(count)]
    return json.dumps({"uuids": uuids, "count": len(uuids)}, indent=2)

import platform
 
 
@mcp.resource("config://server")
def get_server_config() -> str:
    """当前 Server 的配置信息。"""
    config = {
        "name": "my-mcp-server",
        "version": "1.0.0",
        "python_version": platform.python_version(),
        "platform": platform.system(),
        "tools": ["text_stats", "get_weather", "generate_uuid"]
    }
    return json.dumps(config, indent=2)
 
 
# ========== 动态资源模板 ==========
@mcp.resource("users://{user_id}/profile")
def get_user_profile(user_id: str) -> str:
    """用户资料信息。
 
    Args:
        user_id: 用户 ID
    """
    profiles = {
        "user-001": {"name": "张三", "role": "开发者", "team": "后端"},
        "user-002": {"name": "李四", "role": "设计师", "team": "产品"},
        "user-003": {"name": "王五", "role": "运维", "team": "基础设施"},
    }
 
    profile = profiles.get(user_id)
    if not profile:
        raise ValueError(f"用户 {user_id} 不存在")
 
    return json.dumps(profile, indent=2, ensure_ascii=False)

from mcp.server.fastmcp.prompts import base
 
 
@mcp.prompt()
def code_review(
    code: str,
    language: str = "auto",
    focus: str = "all"
) -> list[base.Message]:
    """对代码进行专业审查，提供改进建议。
 
    Args:
        code: 要审查的代码
        language: 编程语言（auto 表示自动检测）
        focus: 审查重点（security/performance/readability/all）
    """
    focus_map = {
        "security": "安全漏洞（SQL 注入、XSS、认证绕过等）",
        "performance": "性能问题（时间复杂度、内存泄漏、N+1 查询等）",
        "readability": "可读性（命名规范、代码结构、注释质量等）",
        "all": "安全性、性能和可读性",
    }
 
    lang_hint = f" {language} " if language != "auto" else ""
    code_block = f"```{language if language != 'auto' else ''}\n{code}\n```"
 
    prompt_text = f"""请对以下{lang_hint}代码进行专业审查。
 
## 审查重点
{focus_map.get(focus, focus_map["all"])}
 
## 代码
{code_block}
 
## 请按以下格式输出
1. **总体评价**：代码质量评分（1-10）和一句话总结
2. **问题列表**：按严重程度排序，每个问题包含位置、描述、修复建议
3. **改进建议**：3-5 条具体的改进建议
4. **优点**：代码中做得好的地方"""
 
    return [base.UserMessage(content=prompt_text)]
 
 
@mcp.prompt()
def generate_sql(
    description: str,
    dialect: str = "postgresql",
    tables: str = ""
) -> list[base.Message]:
    """根据自然语言描述生成 SQL 查询。
 
    Args:
        description: 用自然语言描述你想查询的数据
        dialect: SQL 方言（postgresql/mysql/sqlite）
        tables: 相关表结构描述（可选）
    """
    messages: list[base.Message] = [
        base.AssistantMessage(
            content=f"我是一个 SQL 专家，擅长 {dialect.upper()} 方言。"
            "我会根据你的描述生成安全、高效的 SQL 查询。"
        )
    ]
 
    user_text = f"请根据以下描述生成 {dialect.upper()} 查询：\n\n{description}"
    if tables:
        user_text += f"\n\n相关表结构：\n{tables}"
    user_text += "\n\n请提供：\n1. SQL 查询语句\n2. 查询说明\n3. 性能优化建议（如需要）"
 
    messages.append(base.UserMessage(content=user_text))
    return messages

#!/usr/bin/env python3
"""一个完整的示例 MCP Server。"""
 
import json
import uuid
import platform
from datetime import datetime, timezone
 
from mcp.server.fastmcp import FastMCP
from mcp.server.fastmcp.prompts import base
 
# 创建 Server
mcp = FastMCP(
    name="my-mcp-server",
    version="1.0.0",
    description="一个示例 MCP Server，展示工具、资源和 Prompt 模板"
)
 
# --- Tools ---
@mcp.tool()
def text_stats(text: str, include_whitespace: bool = False) -> str:
    """分析文本的统计信息。"""
    lines = len(text.split("\n"))
    words = len(text.split())
    chars = len(text) if include_whitespace else len(text.replace(" ", ""))
    return json.dumps({"lines": lines, "words": words, "chars": chars}, indent=2)
 
@mcp.tool()
def get_weather(city: str, unit: str = "celsius") -> str:
    """获取指定城市的当前天气信息。"""
    data = {"city": city, "temperature": 25, "condition": "晴", "unit": unit}
    return json.dumps(data, indent=2, ensure_ascii=False)
 
@mcp.tool()
def generate_uuid(count: int = 1) -> str:
    """生成指定数量的 UUID v4。"""
    uuids = [str(uuid.uuid4()) for _ in range(count)]
    return json.dumps({"uuids": uuids}, indent=2)
 
# --- Resources ---
@mcp.resource("config://server")
def get_server_config() -> str:
    """Server 配置信息。"""
    return json.dumps({"name": "my-mcp-server", "version": "1.0.0"}, indent=2)
 
@mcp.resource("users://{user_id}/profile")
def get_user_profile(user_id: str) -> str:
    """用户资料。"""
    profiles = {"user-001": {"name": "张三"}, "user-002": {"name": "李四"}}
    return json.dumps(profiles.get(user_id, {"error": "未找到"}), ensure_ascii=False)
 
# --- Prompts ---
@mcp.prompt()
def code_review(code: str, language: str = "auto") -> list[base.Message]:
    """代码审查模板。"""
    return [base.UserMessage(content=f"请审查以下代码：\n```{language}\n{code}\n```")]
 
# 启动
if __name__ == "__main__":
    mcp.run()

# 方式 1：直接运行
python server.py
 
# 方式 2：使用 uv 运行
uv run server.py
 
# 方式 3：指定传输方式
python server.py --transport stdio        # stdio（默认）
python server.py --transport sse           # SSE（旧版兼容）
python server.py --transport streamable-http  # Streamable HTTP

# 直接通过 npx 启动（无需安装）
npx @modelcontextprotocol/inspector

我的 MCP Server 在 Inspector 中连接失败。请帮我排查：

环境信息：
- 语言：[TypeScript / Python]
- 运行命令：[具体命令]
- 错误信息：[Inspector 显示的错误]

请检查：
1. Server 是否正确输出 JSON-RPC 消息到 stdout
2. 是否有日志或调试信息混入 stdout
3. 依赖是否正确安装
4. 传输方式配置是否正确

{
  "mcpServers": {
    "my-mcp-server": {
      "command": "node",
      "args": ["/absolute/path/to/my-mcp-server/dist/index.js"]
    }
  }
}

{
  "mcpServers": {
    "my-mcp-server": {
      "command": "python",
      "args": ["/absolute/path/to/my-mcp-server/server.py"]
    }
  }
}

{
  "mcpServers": {
    "my-mcp-server": {
      "command": "uv",
      "args": ["run", "--directory", "/absolute/path/to/my-mcp-server", "server.py"]
    }
  }
}

# 添加 MCP Server（TypeScript）
claude mcp add my-mcp-server node /absolute/path/to/dist/index.js
 
# 添加 MCP Server（Python）
claude mcp add my-mcp-server python /absolute/path/to/server.py
 
# 查看已配置的 Server
claude mcp list
 
# 移除 Server
claude mcp remove my-mcp-server

{
  "servers": {
    "my-mcp-server": {
      "command": "node",
      "args": ["./dist/index.js"],
      "cwd": "/absolute/path/to/my-mcp-server"
    }
  }
}

{
  "mcpServers": {
    "my-mcp-server": {
      "command": "node",
      "args": ["/path/to/dist/index.js"],
      "env": {
        "API_KEY": "your-api-key-here",
        "DATABASE_URL": "postgresql://localhost:5432/mydb"
      }
    }
  }
}

server.tool(
  "greet",
  "向用户打招呼",
  { name: z.string().describe("用户名") },
  async ({ name }) => ({
    content: [{ type: "text" as const, text: `你好，${name}！` }]
  })
);

@mcp.tool()
def greet(name: str) -> str:
    """向用户打招呼。
 
    Args:
        name: 用户名
    """
    return f"你好，{name}！"

server.resource(
  "status",
  "status://server",
  { description: "服务器状态" },
  async (uri) => ({
    contents: [{ uri: uri.href, mimeType: "text/plain", text: "running" }]
  })
);

@mcp.resource("status://server")
def get_status() -> str:
    """服务器状态。"""
    return "running"

server.prompt(
  "summarize",
  "总结文本",
  { text: z.string().describe("要总结的文本") },
  async ({ text }) => ({
    messages: [{
      role: "user" as const,
      content: { type: "text" as const, text: `请总结：\n${text}` }
    }]
  })
);

@mcp.prompt()
def summarize(text: str) -> list[base.Message]:
    """总结文本。"""
    return [base.UserMessage(content=f"请总结：\n{text}")]

#!/usr/bin/env python3
"""项目管理 MCP Server：管理待办事项。"""
 
import json
from datetime import datetime, timezone
from dataclasses import dataclass, asdict
from enum import Enum
 
from mcp.server.fastmcp import FastMCP
from mcp.server.fastmcp.prompts import base
 
mcp = FastMCP(name="todo-server", version="1.0.0")
 
 
class Priority(str, Enum):
    LOW = "low"
    MEDIUM = "medium"
    HIGH = "high"
 
 
@dataclass
class Todo:
    id: str
    title: str
    description: str = ""
    priority: str = "medium"
    completed: bool = False
    created_at: str = ""
 
    def __post_init__(self):
        if not self.created_at:
            self.created_at = datetime.now(timezone.utc).isoformat()
 
 
# 内存存储（生产环境应使用数据库）
todos: dict[str, Todo] = {}
_counter = 0
 
 
def _next_id() -> str:
    global _counter
    _counter += 1
    return f"todo-{_counter:04d}"
 
 
# ===== Tools =====
 
@mcp.tool()
def add_todo(title: str, description: str = "", priority: str = "medium") -> str:
    """创建一个新的待办事项。
 
    Args:
        title: 待办事项标题
        description: 详细描述（可选）
        priority: 优先级（low/medium/high）
    """
    todo_id = _next_id()
    todo = Todo(id=todo_id, title=title, description=description, priority=priority)
    todos[todo_id] = todo
    return json.dumps({"message": f"已创建待办事项", "todo": asdict(todo)},
                       indent=2, ensure_ascii=False)
 
 
@mcp.tool()
def complete_todo(todo_id: str) -> str:
    """将待办事项标记为已完成。
 
    Args:
        todo_id: 待办事项 ID
    """
    if todo_id not in todos:
        raise ValueError(f"待办事项 {todo_id} 不存在")
    todos[todo_id].completed = True
    return json.dumps({"message": "已标记完成", "todo": asdict(todos[todo_id])},
                       indent=2, ensure_ascii=False)
 
 
@mcp.tool()
def delete_todo(todo_id: str) -> str:
    """删除一个待办事项。
 
    Args:
        todo_id: 待办事项 ID
    """
    if todo_id not in todos:
        raise ValueError(f"待办事项 {todo_id} 不存在")
    deleted = todos.pop(todo_id)
    return json.dumps({"message": "已删除", "deleted": asdict(deleted)},
                       indent=2, ensure_ascii=False)
 
 
# ===== Resources =====
 
@mcp.resource("todos://all")
def list_all_todos() -> str:
    """所有待办事项列表。"""
    return json.dumps(
        [asdict(t) for t in todos.values()],
        indent=2, ensure_ascii=False
    )
 
 
@mcp.resource("todos://stats")
def todo_stats() -> str:
    """待办事项统计信息。"""
    total = len(todos)
    completed = sum(1 for t in todos.values() if t.completed)
    by_priority = {}
    for t in todos.values():
        by_priority[t.priority] = by_priority.get(t.priority, 0) + 1
    return json.dumps({
        "total": total, "completed": completed,
        "pending": total - completed, "by_priority": by_priority
    }, indent=2)
 
 
@mcp.resource("todos://{todo_id}")
def get_todo(todo_id: str) -> str:
    """单个待办事项详情。"""
    if todo_id not in todos:
        raise ValueError(f"待办事项 {todo_id} 不存在")
    return json.dumps(asdict(todos[todo_id]), indent=2, ensure_ascii=False)
 
 
# ===== Prompts =====
 
@mcp.prompt()
def plan_day(tasks_context: str = "") -> list[base.Message]:
    """根据当前待办事项规划今日工作。
 
    Args:
        tasks_context: 额外的上下文信息（如会议安排、截止日期等）
    """
    todo_list = [asdict(t) for t in todos.values() if not t.completed]
    prompt = f"""请根据以下待办事项帮我规划今天的工作：
 
## 未完成的待办事项
{json.dumps(todo_list, indent=2, ensure_ascii=False)}
 
{f"## 额外上下文{chr(10)}{tasks_context}" if tasks_context else ""}
 
请按以下格式输出：
1. **今日重点**：最重要的 3 件事
2. **时间安排**：按时间段分配任务
3. **建议**：提高效率的建议"""
 
    return [base.UserMessage(content=prompt)]
 
 
if __name__ == "__main__":
    mcp.run()