MCP开发-06.进阶示例-Resources与Tasks

发表于 | 分类于
可运行的进阶 MCP Server 示例：同步 Tool、Resource、Prompt，以及带 Progress 的长任务 Tool（Tasks 需 FastMCP 2.14+ 与 Host 支持）。

本系列：00 什么是 MCP · 01 协议与心智模型 · 02 Python 框架选型 · 03 部署与调用 · 04 重任务与优化 · 05 从 0 开发 · 06 进阶示例（本文） · 07 阿里云案例

1. 示例目标

在 05 最小 Tool 基础上，同一 Server 演示：

原语 示例能力
Tool 同步查询：统计字符串词数
Resource 只读配置 config://app
Prompt 参数化代码审查模板
Tool + Task 模拟批处理（Progress + 后台 Task）

完整源码见同目录 advanced_server.py。

2. Host 侧：Tool vs Resource vs Prompt

原语 模型/Host 典型行为 你何时主动提
Tool 模型在需要「执行动作」时 tools/call 「用 xxx 工具统计词数」
Resource Host 或模型 resources/read 拉上下文 「读取 config 资源」；部分 Host 需手动订阅
Prompt 用户或模型选用模板展开为消息 「使用 code_review 提示」

Resources 不应替代 Tool 做有副作用操作；大结果应用 Resource 卸载（见 04）。

3. 完整示例代码

以下以 官方 SDK 的 FastMCP 为主；§3.4 长任务 需 FastMCP 2.14+（pip install "fastmcp>=2.14"）且 Inspector / Host 支持 Tasks。若环境仅有 mcp[cli]，可注释 §3.4，改用 04 的三 Tool 兼容模式。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
"""进阶 MCP Server：Tools + Resources + Prompts +（可选）Tasks。"""
from __future__ import annotations

import asyncio
import json
from typing import Any

from mcp.server.fastmcp import Context, FastMCP

APP_CONFIG: dict[str, Any] = {
    "app_name": "mcp-advanced-demo",
    "version": "1.0.0",
    "max_batch_chunks": 10,
}

mcp = FastMCP("advanced_demo")


# --- Tool：同步 ---

@mcp.tool()
def count_words(text: str) -> dict[str, int]:
    """统计文本词数（空白分词，简单演示）。

    Args:
        text: 输入文本

    Returns:
        含 word_count 的字典
    """
    words = [w for w in text.split() if w.strip()]
    return {"word_count": len(words)}


# --- Resource：只读 ---

@mcp.resource("config://app")
def app_config() -> str:
    """应用只读配置（JSON）。"""
    return json.dumps(APP_CONFIG, ensure_ascii=False, indent=2)


# --- Prompt：模板 ---

@mcp.prompt()
def code_review(language: str, focus: str = "security") -> str:
    """生成代码审查提示模板。

    Args:
        language: 编程语言
        focus: 审查重点，如 security / performance
    """
    return (
        f"请对以下 {language} 代码进行审查，重点关注 {focus}。"
        "列出问题、风险等级与修改建议。"
    )


# --- Tool + Task：长任务（需 fastmcp 2.14+ 与 Host Tasks 支持）---
# 官方 SDK 跟进 Tasks 前，可改用独立包：
#   from fastmcp import FastMCP
# 并确保 @mcp.tool(task=True) 可用。

@mcp.tool(task=True)
async def batch_process(chunks: int, ctx: Context) -> str:
    """模拟分块批处理（后台 Task + 进度）。

    Args:
        chunks: 分块数量，1–10

    Returns:
        完成摘要
    """
    n = max(1, min(chunks, APP_CONFIG["max_batch_chunks"]))
    for i in range(n):
        await asyncio.sleep(0.3)
        await ctx.report_progress(i + 1, n, f"chunk {i + 1}/{n}")
    return f"processed {n} chunks"


if __name__ == "__main__":
    import sys

    transport = "stdio"
    if len(sys.argv) > 1 and sys.argv[1] == "--http":
        transport = "streamable-http"
    mcp.run(transport=transport)

3.1 运行方式

1
2
3
4
5
6
uv init advanced-mcp && cd advanced-mcp
uv add "mcp[cli]"
# 长任务 Task 需额外：uv add "fastmcp>=2.14"
# 复制 advanced_server.py 后：
uv run python advanced_server.py          # stdio，供 Host 配置
uv run python advanced_server.py --http   # 本地 HTTP + Inspector

3.2 stdio Host 配置（片段）

1
2
3
4
5
6
7
8
{
  "mcpServers": {
    "advanced_demo": {
      "command": "/abs/path/to/advanced-mcp/.venv/bin/python",
      "args": ["/abs/path/to/advanced-mcp/advanced_server.py"]
    }
  }
}

3.3 Inspector 验证清单

  1. 连接后 tools/list 应含 count_words、batch_process。
  2. resources/list 应含 config://app；读取得到 JSON。
  3. prompts/list 应含 code_review。
  4. 调用 batch_process：若返回 Task（status: working），在 Inspector 中轮询 tasks/get 直至 completed；若 Host 不支持 Tasks，可能超时或报错 → 见 §5。

4. Task 生命周期（Inspector / 支持 Tasks 的 Host）

1
2
3
4
5
6
7
8
9
10
11
12
sequenceDiagram
    participant I as Inspector
    participant S as Server

    I->>S: tools/call batch_process
    S-->>I: Task taskId status=working
    loop poll
        I->>S: tasks/get
        S-->>I: working + progress
    end
    I->>S: tasks/get
    S-->>I: completed + result

与 01 Tasks 小节 及 04 重任务 一致。

5. Host 不支持 Tasks 时的降级

保留 count_words、app_config、code_review；将长任务拆为三个同步 Tool（内存 dict 或 Redis 存 job 状态）：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
_JOBS: dict[str, str] = {}

@mcp.tool()
def submit_batch(chunks: int) -> str:
    """提交批处理，返回 job_id（兼容无 Tasks 的 Host）。"""
    job_id = f"job-{len(_JOBS) + 1}"
    _JOBS[job_id] = "working"
    # 生产：推 Celery；此处仅演示
    _JOBS[job_id] = "completed"
    return job_id

@mcp.tool()
def batch_status(job_id: str) -> str:
    """查询 job 状态。"""
    return _JOBS.get(job_id, "unknown")

在 Tool description 中写清调用顺序。

6. 与部署、重任务的衔接

话题 篇章
stdio vs HTTP、Docker、Serverless 03 部署与调用
队列、Resource 卸载、取消 04 重服务与长任务
框架与 uv 环境 02 Python 框架选型
阿里云 Argo + 多 MCP 蛋白设计 07 阿里云案例

7. 小结

能力 装饰器 / API
Tool @mcp.tool()
Resource @mcp.resource("uri://...")
Prompt @mcp.prompt()
长任务 @mcp.tool(task=True) + Context.report_progress

版本提示：Tasks 与 task=True 依赖 FastMCP ≥2.14 及 Client 的 Tasks 实现；以 官方 SDK release 为准，逐步迁移至单一 mcp 包。

-------------本文结束感谢您的阅读-------------