MCP Server 如何实现资源订阅与实时通知？从文件监听到多端广播的完整实战
MCP Server 如何实现资源订阅与实时通知？从文件监听到多端广播的完整实战
本文带你从零搭建一个支持资源订阅（Resource Subscription）与实时通知（Notification）的 MCP Server，涵盖协议原理、Python SDK 实现、文件监听集成、多客户端广播以及 Docker 生产部署，所有代码均可直接复制运行。
一、引言：为什么 MCP 需要资源订阅与通知机制
在传统的 MCP 交互模式中，客户端通过 resources/list 发现资源、通过 resources/read 读取资源内容。这是一种典型的"请求-响应"模型：客户端主动拉取，服务器被动返回。然而，在真实业务场景中，数据往往是动态变化的——日志文件持续追加、数据库记录被修改、配置文件被更新。如果客户端只能轮询，不仅浪费带宽，还会引入不可接受的延迟。
MCP 协议从设计之初就考虑了这个问题。它在 Resources 规范中引入了两个关键能力：
subscribe：允许客户端订阅特定资源 URI，当该资源内容发生变化时，服务器主动推送通知。listChanged：当可用资源列表本身发生变化时（如新增或删除资源），服务器发送列表变更通知。
这套机制将 MCP 从"同步拉取"升级为"推送驱动"，使其能够支撑实时数据推送场景，例如：AI 助手实时感知代码仓库变更、运维大屏秒级更新监控指标、多人协作场景下的状态同步等。
二、MCP 通知协议基础
2.1 能力声明
服务器在初始化握手时，通过 capabilities 字段声明是否支持资源订阅：
{
"capabilities": {
"resources": {
"subscribe": true,
"listChanged": true
}
}
}subscribe 和 listChanged 是互相独立的可选能力——服务器可以只支持其中一个，或都不支持。
2.2 订阅请求与更新通知
客户端通过 resources/subscribe 订阅指定 URI 的资源：
{
"jsonrpc": "2.0",
"id": 4,
"method": "resources/subscribe",
"params": {
"uri": "file:///project/logs/app.log"
}
}当资源内容变化时，服务器发送 notifications/resources/updated 通知：
{
"jsonrpc": "2.0",
"method": "notifications/resources/updated",
"params": {
"uri": "file:///project/logs/app.log"
}
}关键设计理念：通知只携带 URI，不携带资源内容本身。客户端收到通知后，需要再次调用 resources/read 拉取最新内容。这是一种"事件驱动 + 按需拉取"的混合模式，既避免了推送大量数据，又保证了客户端能获取到最新状态。
2.3 资源列表变更通知
当服务器新增或删除资源时，会发送 notifications/resources/list_changed：
{
"jsonrpc": "2.0",
"method": "notifications/resources/list_changed"
}客户端收到后应重新调用 resources/list 刷新资源目录。
三、搭建带订阅功能的 MCP Server
3.1 环境准备
# 创建项目目录
mkdir mcp-sub-demo && cd mcp-sub-demo
# 创建虚拟环境
python -m venv .venv
source .venv/bin/activate # Windows: .venv\Scripts\activate
# 安装依赖
pip install "mcp[cli]" watchdog redis3.2 核心 Server 实现
以下是一个完整的 MCP Server，它暴露日志文件资源，并支持客户端订阅变更通知：
# server.py
import asyncio
import json
from datetime import datetime
from pathlib import Path
from mcp.server.fastmcp import FastMCP, Context
# 初始化 MCP Server
mcp = FastMCP("LogWatcher")
# 模拟日志目录
LOG_DIR = Path("./logs")
LOG_DIR.mkdir(exist_ok=True)
# 默认日志文件
DEFAULT_LOG = LOG_DIR / "app.log"
DEFAULT_LOG.write_text(f"[{datetime.now().isoformat()}] Server started\n", encoding="utf-8")
# 全局订阅管理：记录每个 URI 对应的活跃 session
_subscribers: dict[str, set] = {}
@mcp.resource("log://{filename}")
def read_log(filename: str) -> str:
"""读取指定日志文件内容"""
log_path = LOG_DIR / filename
if not log_path.exists():
raise FileNotFoundError(f"Log file not found: {filename}")
return log_path.read_text(encoding="utf-8")
@mcp.tool()
async def append_log(filename: str, message: str, ctx: Context) -> str:
"""向日志文件追加一条记录，并通知所有订阅者"""
log_path = LOG_DIR / filename
timestamp = datetime.now().isoformat()
line = f"[{timestamp}] {message}\n"
with open(log_path, "a", encoding="utf-8") as f:
f.write(line)
# 核心：通知所有订阅了该资源 URI 的客户端
await ctx.session.send_resource_updated(f"log://{filename}")
return f"Appended to {filename}: {message}"
@mcp.tool()
async def create_log(filename: str, ctx: Context) -> str:
"""创建新日志文件，并通知资源列表变更"""
log_path = LOG_DIR / filename
if log_path.exists():
return f"File already exists: {filename}"
log_path.write_text(f"[{datetime.now().isoformat()}] Created\n", encoding="utf-8")
# 通知资源列表已变更，客户端应重新拉取 resources/list
await ctx.session.send_resource_list_changed()
return f"Created new log file: {filename}"
if __name__ == "__main__":
# 以 stdio 模式运行（适合本地开发与 IDE 集成）
mcp.run(transport="stdio")3.3 关键 API 解析
@mcp.resource("log://{filename}")：注册一个参数化资源模板，客户端可以通过log://app.log这样的 URI 读取对应文件。ctx.session.send_resource_updated(uri)：向所有订阅了该 URI 的客户端发送notifications/resources/updated通知。这是实现实时推送的核心方法。ctx.session.send_resource_list_changed()：通知客户端资源列表已变更，客户端应重新调用resources/list。
四、客户端订阅与实时更新实现
4.1 异步客户端
以下客户端连接到 MCP Server，订阅日志资源，并在收到变更通知时自动拉取最新内容：
# client.py
import asyncio
from mcp import ClientSession, StdioServerParameters
from mcp.client.stdio import stdio_client
async def main():
# 通过 stdio 连接 MCP Server
server_params = StdioServerParameters(
command="python",
args=["server.py"],
)
async with stdio_client(server_params) as (read_stream, write_stream):
async with ClientSession(read_stream, write_stream) as session:
# 初始化连接
await session.initialize()
print("Connected to MCP Server")
# 列出可用资源
resources = await session.list_resources()
print(f"Available resources: {[r.uri for r in resources.resources]}")
# 订阅指定日志资源
target_uri = "log://app.log"
await session.subscribe_resource(target_uri)
print(f"Subscribed to: {target_uri}")
# 初始读取
result = await session.read_resource(target_uri)
print(f"Initial content:\n{result.contents[0].text}")
print("\nWaiting for updates... (try calling append_log tool)\n")
# 监听通知
async for message in session.incoming_messages:
# 检查是否为资源更新通知
if hasattr(message, "method"):
if message.method == "notifications/resources/updated":
uri = message.params.uri
print(f"\n[UPDATE] Resource changed: {uri}")
# 收到通知后重新拉取最新内容
result = await session.read_resource(uri)
print(f"Latest content:\n{result.contents[0].text}")
elif message.method == "notifications/resources/list_changed":
print("\n[LIST CHANGED] Resource list updated")
resources = await session.list_resources()
print(f"Updated resources: {[r.uri for r in resources.resources]}")
if __name__ == "__main__":
asyncio.run(main())4.2 使用 HTTP/SSE 传输模式
在生产环境中，Server 通常以 HTTP 模式运行，支持多客户端同时连接：
# server_http.py —— 替换 server.py 末尾的运行方式
if __name__ == "__main__":
mcp.run(transport="sse", host="0.0.0.0", port=8000)对应客户端使用 SSE 连接：
# client_sse.py
import asyncio
from mcp.client.sse import sse_client
from mcp import ClientSession
async def main():
async with sse_client("http://localhost:8000/sse") as (read_stream, write_stream):
async with ClientSession(read_stream, write_stream) as session:
await session.initialize()
# 订阅并监听
await session.subscribe_resource("log://app.log")
print("Subscribed via SSE, waiting for updates...")
async for message in session.incoming_messages:
if hasattr(message, "method") and \
message.method == "notifications/resources/updated":
uri = message.params.uri
result = await session.read_resource(uri)
print(f"[UPDATE] {uri}:\n{result.contents[0].text}")
if __name__ == "__main__":
asyncio.run(main())五、文件监听与数据库变更推送实战
5.1 watchdog 文件变更监听器
在实际场景中，日志文件可能被外部进程（如应用服务）写入。我们需要用 watchdog 监听文件系统变化，自动触发 MCP 资源更新通知：
# file_watcher.py
import asyncio
import threading
from pathlib import Path
from watchdog.observers import Observer
from watchdog.events import FileSystemEventHandler
from mcp.server.fastmcp import FastMCP, Context
mcp = FastMCP("FileWatcher")
WATCH_DIR = Path("./watched")
WATCH_DIR.mkdir(exist_ok=True)
# 保存当前事件循环引用，供 watchdog 线程回调中使用
_loop: asyncio.AbstractEventLoop | None = None
class LogChangeHandler(FileSystemEventHandler):
"""文件变更事件处理器"""
def on_modified(self, event):
if event.is_directory:
return
self._notify(event.src_path)
def on_created(self, event):
if event.is_directory:
return
self._notify(event.src_path)
def _notify(self, file_path: str):
"""将文件变更转化为 MCP 资源更新通知"""
if _loop is None:
return
# 将 URI 构造为 MCP 资源格式
filename = Path(file_path).name
uri = f"file://{file_path}"
# 从 watchdog 线程向事件循环提交通知任务
asyncio.run_coroutine_threadsafe(
_broadcast_update(uri), _loop
)
async def _broadcast_update(uri: str):
"""向所有活跃 session 广播资源更新通知"""
from mcp.server.session import ServerSession
# 遍历所有活跃的 session 并发送通知
for session in mcp._mcp_server.sessions:
try:
await session.send_resource_updated(uri)
except Exception as e:
print(f"Failed to notify session: {e}")
def start_watcher():
"""在独立线程中启动文件监听"""
observer = Observer()
observer.schedule(LogChangeHandler(), str(WATCH_DIR), recursive=True)
observer.start()
return observer
@mcp.resource("file://{path}")
def read_file(path: str) -> str:
"""读取被监听目录中的文件"""
file_path = WATCH_DIR / path
if not file_path.exists():
raise FileNotFoundError(f"File not found: {path}")
return file_path.read_text(encoding="utf-8")
@mcp.tool()
async def write_file(path: str, content: str, ctx: Context) -> str:
"""写入文件内容（会触发 watchdog 自动通知）"""
file_path = WATCH_DIR / path
file_path.write_text(content, encoding="utf-8")
# watchdog 会自动捕获变更并发送通知
# 也可以手动通知以确保即时性
await ctx.session.send_resource_updated(f"file://{file_path}")
return f"Written {len(content)} bytes to {path}"
@mcp.tool()
async def list_watched_files(ctx: Context) -> str:
"""列出所有被监听的文件"""
files = [f.name for f in WATCH_DIR.iterdir() if f.is_file()]
return json.dumps(files) if files else "No files found"
import json
if __name__ == "__main__":
# 捕获事件循环引用
_loop = asyncio.new_event_loop()
asyncio.set_event_loop(_loop)
# 启动文件监听线程
watcher = start_watcher()
print(f"Watching directory: {WATCH_DIR.resolve()}")
try:
mcp.run(transport="sse", host="0.0.0.0", port=8000)
finally:
watcher.stop()
watcher.join()5.2 数据库变更推送（PostgreSQL LISTEN/NOTIFY 示例）
# db_watcher.py —— 数据库变更监听片段
import asyncio
import asyncpg
from mcp.server.fastmcp import FastMCP, Context
mcp = FastMCP("DBWatcher")
DB_DSN = "postgresql://user:pass@localhost:5432/mydb"
async def listen_db_changes():
"""监听 PostgreSQL 的 NOTIFY 事件并转发为 MCP 资源通知"""
conn = await asyncpg.connect(DB_DSN)
await conn.add_listener("record_changed", _on_db_notification)
# 保持连接
await asyncio.Event().wait()
def _on_db_notification(connection, pid, channel, payload):
"""数据库变更回调：payload 格式为 JSON {"table": "orders", "id": 42}"""
import json
data = json.loads(payload)
uri = f"db://{data['table']}/{data['id']}"
# 向事件循环提交通知任务
asyncio.run_coroutine_threadsafe(
_broadcast_update(uri), asyncio.get_event_loop()
)
@mcp.resource("db://{table}/{record_id}")
async def read_record(table: str, record_id: str) -> str:
"""读取数据库记录"""
conn = await asyncpg.connect(DB_DSN)
row = await conn.fetchrow(
f"SELECT * FROM {table} WHERE id = $1", int(record_id)
)
await conn.close()
if row is None:
raise ValueError(f"Record not found: {table}/{record_id}")
return json.dumps(dict(row), default=str)六、多客户端广播与状态同步
当多个客户端同时订阅同一资源时，服务器需要确保所有订阅者都能收到更新通知。MCP SDK 在内部维护了每个 session 的订阅列表，调用 send_resource_updated 时会自动向所有匹配的 session 广播。
6.1 广播架构
┌──────────────┐
文件变更/DB事件 ──→│ 事件源 │
└──────┬───────┘
│ publish
┌──────▼───────┐
│ SubscriptionBus│ ← 内存总线（单进程）
│ 或 Redis Bus │ ← 分布式总线（多进程）
└──┬───┬───┬───┘
│ │ │
┌────────┘ │ └────────┐
▼ ▼ ▼
┌──────────┐ ┌──────────┐ ┌──────────┐
│ Client A │ │ Client B │ │ Client C │
│ Session 1 │ │ Session 2 │ │ Session 3 │
└──────────┘ └──────────┘ └──────────┘6.2 基于 Redis 的跨进程广播
当 Server 以多副本部署时，单进程内存总线无法跨进程传递通知。此时需要使用 Redis Pub/Sub 作为消息总线：
# redis_bus.py —— 跨进程订阅总线
import asyncio
import json
from collections.abc import Callable
from redis.asyncio import Redis
class RedisSubscriptionBus:
"""基于 Redis Pub/Sub 的跨进程订阅事件总线"""
CHANNEL = "mcp:resource:updates"
def __init__(self, redis_url: str = "redis://localhost:6379"):
self._redis = Redis.from_url(redis_url)
self._listeners: dict[object, Callable] = {}
self._reader_task: asyncio.Task | None = None
async def start(self):
"""启动 Redis 订阅读取任务"""
self._reader_task = asyncio.create_task(self._read_loop())
async def stop(self):
"""停止读取任务"""
if self._reader_task:
self._reader_task.cancel()
await self._redis.close()
async def _read_loop(self):
"""持续监听 Redis 频道，将消息分发给本地监听器"""
pubsub = self._redis.pubsub()
await pubsub.subscribe(self.CHANNEL)
try:
async for message in pubsub.listen():
if message["type"] == "message":
event = json.loads(message["data"])
for listener in list(self._listeners.values()):
try:
listener(event)
except Exception as e:
print(f"Listener error: {e}")
except asyncio.CancelledError:
pass
finally:
await pubsub.unsubscribe(self.CHANNEL)
async def publish(self, uri: str, event_type: str = "updated"):
"""发布资源变更事件到 Redis（所有副本都能收到）"""
event = {"uri": uri, "type": event_type}
await self._redis.publish(self.CHANNEL, json.dumps(event))
def subscribe(self, listener: Callable[[dict], None]) -> Callable[[], None]:
"""注册本地监听器，返回取消订阅函数"""
token = object()
self._listeners[token] = listener
def unsubscribe():
self._listeners.pop(token, None)
return unsubscribe在 Server 中集成 Redis 总线：
# server_distributed.py
import asyncio
from redis_bus import RedisSubscriptionBus
from mcp.server.fastmcp import FastMCP, Context
mcp = FastMCP("DistributedWatcher")
bus = RedisSubscriptionBus("redis://redis:6379")
# 存储 uri -> set[session] 的映射
_local_subs: dict[str, set] = {}
@mcp.tool()
async def update_data(key: str, value: str, ctx: Context) -> str:
"""更新数据并通过 Redis 总线广播"""
uri = f"data://{key}"
# 1. 本地通知当前 session 的订阅者
await ctx.session.send_resource_updated(uri)
# 2. 通过 Redis 广播到其他副本
await bus.publish(uri, "updated")
return f"Updated {key}={value}"
async def setup_bus():
"""启动总线并将远程事件分发给本地 session"""
await bus.start()
def on_remote_event(event: dict):
uri = event["uri"]
# 向本地所有订阅了该 URI 的 session 发送通知
for session in mcp._mcp_server.sessions:
asyncio.run_coroutine_threadsafe(
session.send_resource_updated(uri),
asyncio.get_event_loop(),
)
bus.subscribe(on_remote_event)
if __name__ == "__main__":
loop = asyncio.get_event_loop()
loop.run_until_complete(setup_bus())
mcp.run(transport="sse", host="0.0.0.0", port=8000)七、生产部署：Docker 化与 Nginx 反代
7.1 Dockerfile
# Dockerfile
FROM python:3.12-slim
WORKDIR /app
# 安装系统依赖（watchdog 的 inotify 支持）
RUN apt-get update && apt-get install -y --no-install-recommends \
gcc && \
rm -rf /var/lib/apt/lists/*
# 安装 Python 依赖
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt
# 复制源码
COPY . .
# 创建日志与监听目录
RUN mkdir -p /app/logs /app/watched
# 暴露 SSE 端口
EXPOSE 8000
# 健康检查
HEALTHCHECK --interval=30s --timeout=5s --retries=3 \
CMD python -c "import urllib.request; urllib.request.urlopen('http://localhost:8000/health')" || exit 1
# 启动 Server
CMD ["python", "file_watcher.py"]7.2 requirements.txt
mcp[cli]>=1.2.0
watchdog>=4.0.0
redis>=5.0.0
asyncpg>=0.29.0
uvicorn>=0.30.07.3 docker-compose.yml
# docker-compose.yml
version: "3.9"
services:
mcp-server:
build: .
ports:
- "8000:8000"
volumes:
- ./watched:/app/watched
- ./logs:/app/logs
environment:
- REDIS_URL=redis://redis:6379
depends_on:
redis:
condition: service_healthy
restart: unless-stopped
deploy:
replicas: 2 # 多副本部署，通过 Redis 总线同步
redis:
image: redis:7-alpine
ports:
- "6379:6379"
healthcheck:
test: ["CMD", "redis-cli", "ping"]
interval: 5s
timeout: 3s
retries: 5
restart: unless-stopped
nginx:
image: nginx:alpine
ports:
- "80:80"
volumes:
- ./nginx.conf:/etc/nginx/nginx.conf:ro
depends_on:
- mcp-server
restart: unless-stopped7.4 Nginx 反向代理配置
由于 SSE 需要长连接，Nginx 配置必须关闭缓冲并设置超时：
# nginx.conf
events {
worker_connections 1024;
}
http {
upstream mcp_backend {
# 负载均衡到多个 MCP Server 副本
# 注意：SSE 需要使用 ip_hash 保持连接亲和性
ip_hash;
server mcp-server:8000;
}
server {
listen 80;
server_name _;
# MCP SSE 端点
location /sse {
proxy_pass http://mcp_backend;
proxy_http_version 1.1;
proxy_set_header Host $host;
proxy_set_header X-Real-IP $remote_addr;
proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;
# SSE 关键配置：禁用缓冲，支持长连接
proxy_buffering off;
proxy_cache off;
proxy_set_header Connection '';
# 长超时（SSE 连接持续保持）
proxy_read_timeout 86400s;
proxy_send_timeout 86400s;
chunked_transfer_encoding on;
}
# 普通 HTTP 端点
location / {
proxy_pass http://mcp_backend;
proxy_set_header Host $host;
proxy_set_header X-Real-IP $remote_addr;
}
}
}7.5 一键启动
# 构建并启动所有服务
docker compose up -d --build
# 查看日志
docker compose logs -f mcp-server
# 测试 SSE 连接
curl -N http://localhost/sse八、常见问题 FAQ
Q1：客户端收不到 notifications/resources/updated 通知，可能是什么原因？
A：常见原因有三个：（1）Server 未在 capabilities 中声明 "subscribe": true，客户端无法订阅；（2）客户端未调用 resources/subscribe 就直接等待通知，订阅是主动行为；（3）Server 代码中调用的是 print() 而非 ctx.session.send_resource_updated(uri)，通知必须通过 session 对象发送。建议先用 mcp inspector 工具验证 Server 的能力声明和订阅流程。
Q2：send_resource_updated 和 send_resource_list_changed 有什么区别？
A：send_resource_updated(uri) 通知的是某个具体资源的内容发生了变化，只有订阅了该 URI 的客户端会收到，通知中携带 URI 参数。send_resource_list_changed() 通知的是资源目录本身发生了变化（如新增或删除了资源文件），所有客户端都应重新拉取 resources/list，通知不携带参数。前者对应 notifications/resources/updated，后者对应 notifications/resources/list_changed。
Q3：SSE 传输模式下，Nginx 代理后通知延迟很高或连接断开怎么办？
A：这是 Nginx 默认缓冲导致的。必须在 Nginx 的 location 块中设置 proxy_buffering off; 和 proxy_cache off;，并将 proxy_read_timeout 设置为足够大的值（如 86400s）。同时确保 proxy_set_header Connection ''; 和 proxy_http_version 1.1; 已正确配置。如果使用了 ip_hash，请确认负载均衡策略不会在连接中途将请求路由到不同后端。
Q4：多副本部署时，客户端订阅在副本 A，但变更发生在副本 B，如何解决？
A：这是经典的分布式订阅问题。单进程内存总线无法跨进程传递通知，必须引入外部消息中间件。本文提供的 RedisSubscriptionBus 方案是推荐做法：每个副本将本地变更 publish 到 Redis 频道，所有副本的 reader task 收到消息后向各自的本地 session 发送通知。也可以使用 RabbitMQ Fanout Exchange 或 NATS 等替代方案，核心思路一致。
Q5：通知只携带 URI 不携带内容，客户端每次都要重新 read，这样设计合理吗？
A：这是 MCP 协议的刻意设计。原因有三：（1）资源内容可能很大（如完整日志文件），直接推送会浪费带宽；（2）不同客户端可能需要不同的内容格式（如只读权限的客户端只能看到脱敏数据），统一推送无法适配；（3）分离"通知"与"读取"使得协议更简洁，服务器无需维护每个客户端的内容缓存。这种"事件通知 + 按需拉取"的模式与 Webhook 机制类似，是业界验证过的可靠方案。
Q6：watchdog 线程中如何安全地调用异步的 MCP 通知方法？
A：watchdog 的事件回调运行在独立线程中，不能直接 await 异步方法。正确做法是使用 asyncio.run_coroutine_threadsafe(coro, loop) 将协程提交到主事件循环执行。需要提前保存事件循环引用（loop = asyncio.get_event_loop()），在回调中通过该引用提交任务。详见本文 file_watcher.py 中的 _notify 方法实现。
Q7：资源 URI 使用什么格式？可以自定义 scheme 吗？
A：MCP 支持标准 URI scheme（如 file://、https://、git://），也完全支持自定义 scheme（如本文使用的 log://、db://、data://）。自定义 scheme 时，URI 只需在 Server 内部保持一致即可。客户端通过 resources/list 或 resources/templates/list 发现资源 URI，无需预先知道 scheme。注意 URI 中如果包含特殊字符（如空格），需进行 URL 编码。
九、总结
MCP 的资源订阅与通知机制，将协议从被动的"请求-响应"模式提升为主动的"事件驱动"模式，使其能够胜任实时数据推送场景。本文完整覆盖了从协议原理到生产部署的全链路：
- 协议层：
resources/subscribe+notifications/resources/updated构成了订阅通知的核心闭环，通知只携带 URI 不携带内容，客户端按需拉取。 - Server 实现：通过 Python MCP SDK 的
ctx.session.send_resource_updated(uri)一行代码即可向所有订阅者广播变更。 - 文件监听：集成 watchdog 实现文件系统变更的自动感知，通过
run_coroutine_threadsafe桥接线程与异步事件循环。 - 多客户端广播：SDK 内部自动管理 session 级订阅列表，
send_resource_updated天然支持一对多广播。 - 分布式扩展：通过 Redis Pub/Sub 实现跨进程事件总线，支持多副本部署。
- 生产部署：Docker Compose 一键编排，Nginx 反向代理需特殊配置 SSE 长连接支持。
核心原则始终是：通知是信号，不是数据。客户端收到通知后主动拉取最新内容，这种解耦让系统更简洁、更可扩展。随着 MCP 生态的成熟，资源订阅机制将成为构建实时 AI 上下文管道的标准范式。