Skip to content

6.3 综合实战——小周的毕业项目

从 Stage 1 的集成噩梦到现在,小周走过了完整的学习之路。现在是验收成果的时候了——他要开发一个真正能用的项目管理 MCP Server,作为团队的毕业项目。


实战:项目助手 MCP Server

目标:开发一个 MCP Server,为 AI 工具提供项目管理能力。

功能要求

你的 Server 需要暴露:

Tools(至少 4 个):

  • create-task:创建任务(标题、描述、优先级)
  • list-tasks:列出任务(支持按状态/优先级过滤)
  • update-task:更新任务状态(待办→进行中→已完成)
  • get-project-stats:获取项目统计信息

Resources(至少 2 个):

  • project://stats:项目统计数据
  • project://tasks:任务列表概览

Prompts(至少 1 个):

  • daily-standup:根据当前任务状态生成站会报告

技术要求

  1. 使用 Python(FastMCP)或 TypeScript(McpServer)
  2. 数据存储在内存中(不需要真正的数据库)
  3. 使用 Tool Annotations 标注每个工具
  4. 错误处理:所有输入验证错误作为 Tool Execution Error 返回
  5. 代码不超过 150 行

实战步骤

创建项目
实现 Server
测试
配置 Client
安全检查

Step 1:创建项目

bash
mkdir mcp-project-assistant && cd mcp-project-assistant

Step 2:实现 Server

Python 参考实现
python
# server.py
from mcp.server.fastmcp import FastMCP
from datetime import datetime

mcp = FastMCP("project-assistant")

tasks: dict[int, dict] = {}
_next_id = 1

def next_id():
    global _next_id
    _next_id += 1
    return _next_id - 1

@mcp.tool()
def create_task(title: str, description: str = "", priority: str = "medium") -> str:
    """创建任务。priority 可选:low、medium、high"""
    if priority not in ("low", "medium", "high"):
        raise ValueError(f"无效优先级:{priority}。请使用 low/medium/high。")
    tid = next_id()
    tasks[tid] = {
        "id": tid, "title": title, "description": description,
        "priority": priority, "status": "todo",
        "created": datetime.now().isoformat()
    }
    return f"已创建任务 #{tid}{title}{priority})"

@mcp.tool()
def list_tasks(status: str = "", priority: str = "") -> str:
    """列出任务。可按 status(todo/doing/done)和 priority 过滤"""
    filtered = list(tasks.values())
    if status:
        filtered = [t for t in filtered if t["status"] == status]
    if priority:
        filtered = [t for t in filtered if t["priority"] == priority]
    if not filtered:
        return "没有匹配的任务"
    return "\n".join(
        f"#{t['id']} [{t['status']}] ({t['priority']}) {t['title']}"
        for t in filtered
    )

@mcp.tool()
def update_task(task_id: int, status: str) -> str:
    """更新任务状态。status 可选:todo、doing、done"""
    if task_id not in tasks:
        raise ValueError(f"任务 #{task_id} 不存在。当前有 {len(tasks)} 个任务。")
    if status not in ("todo", "doing", "done"):
        raise ValueError(f"无效状态:{status}。请使用 todo/doing/done。")
    tasks[task_id]["status"] = status
    return f"已将任务 #{task_id} 更新为 {status}"

@mcp.tool()
def get_project_stats() -> str:
    """获取项目统计信息"""
    total = len(tasks)
    todo = sum(1 for t in tasks.values() if t["status"] == "todo")
    doing = sum(1 for t in tasks.values() if t["status"] == "doing")
    done = sum(1 for t in tasks.values() if t["status"] == "done")
    high = sum(1 for t in tasks.values() if t["priority"] == "high")
    return f"共 {total} 个任务:{todo} 待办 / {doing} 进行中 / {done} 已完成 / {high} 高优先级"

@mcp.resource("project://stats")
def stats_resource() -> str:
    return get_project_stats()

@mcp.resource("project://tasks")
def tasks_resource() -> str:
    return list_tasks()

@mcp.prompt()
def daily_standup(focus: str = "") -> str:
    """根据当前任务状态生成站会报告"""
    doing = [t for t in tasks.values() if t["status"] == "doing"]
    todo = [t for t in tasks.values() if t["status"] == "todo" and t["priority"] == "high"]
    result = "## 今日站会\n\n"
    result += "### 进行中\n"
    result += "\n".join(f"- #{t['id']} {t['title']}" for t in doing) or "- 无"
    result += "\n\n### 高优先级待办\n"
    result += "\n".join(f"- #{t['id']} {t['title']}" for t in todo) or "- 无"
    if focus:
        result += f"\n\n### 今日重点\n{focus}"
    return result

mcp.run()
TypeScript 参考实现
typescript
// server.ts
import { McpServer } from "@modelcontextprotocol/sdk/server/mcp.js";
import { StdioServerTransport } from "@modelcontextprotocol/sdk/server/stdio.js";
import { z } from "zod";

const server = new McpServer({ name: "project-assistant", version: "1.0.0" });

interface Task {
  id: number;
  title: string;
  description: string;
  priority: "low" | "medium" | "high";
  status: "todo" | "doing" | "done";
  created: string;
}

const tasks = new Map<number, Task>();
let nextId = 1;

server.tool("create-task", "创建任务。priority 可选:low、medium、high",
  {
    title: z.string().describe("任务标题"),
    description: z.string().optional().describe("任务描述"),
    priority: z.enum(["low", "medium", "high"]).optional().describe("优先级")
  },
  async ({ title, description, priority }) => {
    const p = priority || "medium";
    const id = nextId++;
    tasks.set(id, {
      id, title, description: description || "", priority: p,
      status: "todo", created: new Date().toISOString()
    });
    return { content: [{ type: "text" as const, text: `已创建任务 #${id}:${title}(${p})` }] };
  }
);

server.tool("list-tasks", "列出任务。可按 status 和 priority 过滤",
  {
    status: z.enum(["todo", "doing", "done"]).optional(),
    priority: z.enum(["low", "medium", "high"]).optional()
  },
  async ({ status, priority }) => {
    let filtered = [...tasks.values()];
    if (status) filtered = filtered.filter(t => t.status === status);
    if (priority) filtered = filtered.filter(t => t.priority === priority);
    if (filtered.length === 0) {
      return { content: [{ type: "text" as const, text: "没有匹配的任务" }] };
    }
    const list = filtered.map(t =>
      `#${t.id} [${t.status}] (${t.priority}) ${t.title}`
    ).join("\n");
    return { content: [{ type: "text" as const, text: list }] };
  }
);

server.tool("update-task", "更新任务状态。status 可选:todo、doing、done",
  {
    task_id: z.number().describe("任务 ID"),
    status: z.enum(["todo", "doing", "done"]).describe("新状态")
  },
  async ({ task_id, status }) => {
    const task = tasks.get(task_id);
    if (!task) {
      return {
        content: [{ type: "text" as const, text: `任务 #${task_id} 不存在。当前有 ${tasks.size} 个任务。` }],
        isError: true
      };
    }
    task.status = status;
    return { content: [{ type: "text" as const, text: `已将任务 #${task_id} 更新为 ${status}` }] };
  }
);

server.tool("get-project-stats", "获取项目统计信息", {},
  async () => {
    const all = [...tasks.values()];
    const todo = all.filter(t => t.status === "todo").length;
    const doing = all.filter(t => t.status === "doing").length;
    const done = all.filter(t => t.status === "done").length;
    const high = all.filter(t => t.priority === "high").length;
    return {
      content: [{
        type: "text" as const,
        text: `共 ${all.length} 个任务:${todo} 待办 / ${doing} 进行中 / ${done} 已完成 / ${high} 高优先级`
      }]
    };
  }
);

const transport = new StdioServerTransport();
await server.connect(transport);

Step 3:测试

用 MCP Inspector 测试你的 Server:

bash
npx @modelcontextprotocol/inspector python server.py

在 Inspector 中:

  1. 连接 Server,查看初始化交换
  2. 在 Tools Tab 中查看工具列表
  3. 调用 create-task 创建几个任务
  4. 调用 list-tasks 查看列表
  5. 调用 get-project-stats 查看统计
  6. 测试错误处理:传入无效的 priority 看看返回什么

Step 4:配置到 Claude Desktop

claude_desktop_config.json 中添加你的 Server,然后在 Claude Desktop 中测试自然语言交互:

你:帮我创建三个任务:写文档(高优先级)、修 Bug(中优先级)、重构代码(低优先级)

Claude:(调用 create-task 三次)

你:今天的站会报告是什么?

Claude:(查看 project://stats 资源和任务列表,生成站会报告)

Step 5:安全检查

对照检查清单自查:

  • [ ] 每个工具都有描述吗?
  • [ ] 输入验证是否严格?(优先级只能是 low/medium/high)
  • [ ] 错误信息是否可操作?(「请使用 low/medium/high」)
  • [ ] Resource URI 是否清晰?
  • [ ] Prompt 参数是否合理?

进阶挑战

如果你完成了基础实战,试试这些进阶挑战:

  1. 添加 Elicitation:删除任务前请求用户确认
  2. 添加 Sampling:AI 分析任务描述,自动推荐优先级
  3. 切换到 HTTP 模式:把 Server 部署为远程 HTTP 服务
  4. 写一个 Client:用 Python 或 TypeScript 写一个 Client 命令行工具
  5. 添加持久化:把任务保存到文件或 SQLite

课程总结——小周的毕业

恭喜你完成了 MCP Engineering 课程!

Stage 1:集成噩梦
被 M×N 问题折磨,发现了 MCP——AI 的 USB-C
Stage 2:理解架构
三层架构、三大原语、JSON-RPC 消息流
Stage 3:动手实战
写出第一个 Server 和 Client,学会用 Inspector 调试
Stage 4:安全进阶
OAuth 2.1 授权、Sampling 双向通信、远程部署
Stage 5:技术选型
MCP vs OpenAPI vs Function Calling vs A2A——给团队做了评审
Stage 6:毕业项目 ✅
成功交付了团队可用的项目管理 MCP Server

从「每个工具定制集成」到「一次开发,所有 AI 可用」——小周用了 6 小时,你也可以。

下一步:为你的项目开发一个真正有用的 MCP Server,部署到生产环境,让团队成员都能使用。


← 上一节:设计题 | 目录