用 mcp-agent 构建带日志与进度通知能力的 MCP Server 实战指南-开发者社区

用 mcp-agent 构建带日志与进度通知能力的 MCP Server 实战指南

【免费下载链接】mcp-agentBuild effective agents using Model Context Protocol and simple workflow patterns项目地址: https://gitcode.com/GitHub_Trending/mc/mcp-agent

导读

本篇基于 mcp-agent 仓库中的 Notifications Server 示例（src/mcp_agent/data/examples/mcp_agent_server/notifications/README.md），讲解如何构建一个最小化的 MCP Server，向连接它的上游 MCP 客户端转发服务端日志与进度通知。读完本文，你将掌握notify与notify_progress两类通知工具的声明方式、上游会话的获取与使用、logging_callback客户端回调的接入，以及将该服务器部署到 mcp-agent Cloud 的完整流程，可直接复用到任何需要"服务端主动上报状态"的 MCP 场景。

示例概览：一个服务器、两个通知工具

该示例位于src/mcp_agent/data/examples/mcp_agent_server/notifications/目录，包含三个文件：

文件	作用
`server.py`	以`MCPApp`声明两个工具，通过 SSE 传输启动 MCP Server
`client.py`	最小客户端，连接服务器并触发两个通知工具，在回调中打印服务端日志
`README.md`	运行说明与部署说明（本文档）

其核心目标是演示两类服务端通知能力：

logging 通知：服务端把日志以 MCPnotifications/message形式转发给上游客户端；
非 logging 通知：服务端通过notifications/progress上报任务进度。

两个工具的设计原则是best-effort（尽力而为）且对服务器非阻塞：即使上游会话不可用，也不应拖垮服务器主流程。

运行：一键启动服务器与客户端

按 README 的指引，在示例目录下先启动服务器：

uv run server.py

服务器内部会执行create_mcp_server_for_app(agent_app)并调用run_sse_async()，以 SSE（Server-Sent Events）传输在http://127.0.0.1:8000/sse上提供 MCP 服务。之后另开一个终端连接客户端：

uv run client.py

客户端会依次调用两个工具并输出结果，典型运行流程为：

客户端建立到http://127.0.0.1:8000/sse的会话；
调用notify，把Hello from client以 info 级别日志转发给客户端；
调用notify_progress，发送progress=0.25、message="Quarter"的进度通知；
打印Sent notify + notify_progress。

客户端为何能看到服务端日志

客户端在client.py中自定义了会话工厂_make_session，通过MCPAgentClientSession(..., logging_callback=on_server_log)注册了日志回调：

async def on_server_log(params: LoggingMessageNotificationParams) -> None: level = params.level.upper() name = params.logger or "server" print(f"[SERVER LOG] [{level}] [{name}] {params.data}")

MCPAgentClientSession是 mcp-agent 框架对官方ClientSession的派生（见 mcp_agent_client_session.py），在原有请求/通知机制之上补充了日志回调、sampling 处理与根目录配置支持。会话工厂随后通过gen_client的client_session_factory参数注入，同时客户端还调用了set_logging_level("info")以启用 info 及以上级别的日志上报。回调中对level统一做.upper()归一化、对缺省 logger 名回退为"server"，保证输出格式稳定。

服务端实现：声明两个通知工具

notify：把日志转发给上游客户端

server.py通过@app.tool装饰器声明同步工具notify：

@app.tool(name="notify") def notify( message: str, level: Literal["debug", "info", "warning", "error"] = "info", app_ctx: Optional[AppContext] = None, ) -> str: _app = app_ctx.app if app_ctx else app logger = _app.logger if level == "debug": logger.debug(message) elif level == "warning": logger.warning(message) elif level == "error": logger.error(message) else: logger.info(message) return "ok"

要点解析：

@app.tool是MCPApp提供的声明式工具注册接口，声明后工具会以 FastMCP Tool 形式暴露给远端客户端（注册与校验逻辑见 app.py 中的tool/async_tool与 tool_adapter.py）；
level使用Literal["debug", "info", "warning", "error"]约束合法取值，默认info；
app_ctx: Optional[AppContext]是框架自动注入的应用上下文，允许工具在无上下文时回退到模块级app实例；
通过_app.logger（即MCPApp.logger）输出日志。当服务器作为 MCP Server 运行时，上游会话已绑定到应用上下文，日志事件会以notifications/message上报到客户端——这正是客户端logging_callback收到内容的来源。

notify_progress：上报进度通知

notify_progress是异步工具，演示非日志类通知：

@app.tool(name="notify_progress") async def notify_progress( progress: float = 0.5, message: str | None = "Demo progress", app_ctx: Optional[AppContext] = None, ) -> str: _app = app_ctx.app if app_ctx else app upstream = getattr(_app.context, "upstream_session", None) if upstream is None: _app.logger.warning("No upstream session to notify") return "no-upstream" await upstream.send_progress_notification( progress_token="notifications-demo", progress=progress, message=message ) _app.logger.info("Sent notifications/progress") return "ok"

要点解析：

它不依赖日志系统，而是直接读取app.context.upstream_session，这是框架在 MCP Server 模式下保存的"上游客户端会话"引用；
通过send_progress_notification(progress_token, progress, message)发送 MCP 标准进度通知notifications/progress。该方法在 mcp_agent_client_session.py 中有完整实现，支持total可选参数并在启用了 tracing 时记录progress_token、progress等 span 属性；
上游会话缺失时返回"no-upstream"并仅记录 warning，不抛出异常，体现了"best-effort 且非阻塞"的设计意图。

服务器启动入口

两个工具都注册在同一个MCPApp上，main()中通过app.run()生命周期初始化应用，再用create_mcp_server_for_app将应用暴露为 FastMCP Server：

async def main() -> None: async with app.run() as agent_app: mcp_server = create_mcp_server_for_app(agent_app) await mcp_server.run_sse_async()

create_mcp_server_for_app（见 app_server.py）会为应用创建托管 lifespan，并在启动阶段注册工作流工具与函数声明工具，随后以 SSE 方式监听连接。

工作原理：上游会话与通知的中继链路

从源码结构看，通知的中继遵循一条清晰链路：

客户端建立 SSE 连接后，FastMCP 服务器将该客户端的会话保存为upstream_session，绑定到应用与请求上下文（相关绑定逻辑集中在 app_server.py 的_enter_request_context等函数中）；
服务端工具通过app.context.upstream_session拿到该会话引用；
日志事件由框架日志系统自动转发，进度事件则显式调用send_progress_notification发送；
客户端侧MCPAgentClientSession通过logging_callback接收并展示日志，进度通知则由底层 SDK 分发给监听方。

值得一提的是，app_server.py 中还实现了内部 HTTP 路由/internal/session/by-run/{execution_id}/notify，支持notifications/message与notifications/progress两种方法的透传转发，并带有基于MCP_GATEWAY_TOKEN的可选鉴权与幂等键去重。这为 Temporal 等外部 worker 无法直接持有会话对象的场景提供了"经服务器中转"的备用通道，可视为本示例通知能力的生产级扩展。

可选：部署到 mcp-agent Cloud

README 提供了将该服务器部署到 mcp-agent Cloud 的步骤：

配置密钥：在示例目录下的mcp_agent.secrets.yaml中按需设置 API keys（该文件通常以mcp_agent.secrets.yaml.example为模板复制而来）；
部署：在该目录执行

uv run mcp-agent deploy notifications-demo

连接：使用部署返回的 URL，并在末尾加上/sse作为 MCP 客户端连接端点；在请求 Header 中把 Bearer token 设置为你的 mcp-agent API key。

部署后，MCP 客户端通过 SSE 端点连接该服务器，即可获得与本地运行完全一致的日志与进度通知能力。若需要进一步了解部署细节，可参考 deploy-mcp-server.mdx 与 use-deployed-server.mdx。

实践要点与扩展建议

日志级别语义：notify的level参数与 Python logging 级别一一对应，客户端需先set_logging_level设置阈值，低于阈值的日志不会上报；
进度上报频率：进度通知是单向 fire-and-forget 消息，适合在耗时工具中按阶段上报；send_progress_notification支持total参数表达总量，便于客户端渲染进度条；
健壮性：始终用getattr(..., "upstream_session", None)防御性读取，并在会话缺失时优雅降级（如返回"no-upstream"），保证工具不因通知失败而中断；
复用路径：若要在自己项目中复刻本示例，直接参考 server.py 与 client.py，并把工具体替换为你的业务逻辑即可；更完整的服务器模式样例还可对比src/mcp_agent/data/examples/mcp_agent_server/下的 asyncio、sampling、elicitation 等子示例。

【免费下载链接】mcp-agentBuild effective agents using Model Context Protocol and simple workflow patterns项目地址: https://gitcode.com/GitHub_Trending/mc/mcp-agent

创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考