用 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 Serverclient.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()以 SSEServer-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发送progress0.25、messageQuarter的进度通知打印Sent notify notify_progress。客户端为何能看到服务端日志客户端在client.py中自定义了会话工厂_make_session通过MCPAgentClientSession(..., logging_callbackon_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装饰器声明同步工具notifyapp.tool(namenotify) 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.pylevel使用Literal[debug, info, warning, error]约束合法取值默认infoapp_ctx: Optional[AppContext]是框架自动注入的应用上下文允许工具在无上下文时回退到模块级app实例通过_app.logger即MCPApp.logger输出日志。当服务器作为 MCP Server 运行时上游会话已绑定到应用上下文日志事件会以notifications/message上报到客户端——这正是客户端logging_callback收到内容的来源。notify_progress上报进度通知notify_progress是异步工具演示非日志类通知app.tool(namenotify_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_tokennotifications-demo, progressprogress, messagemessage ) _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 Serverasync 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 CloudREADME 提供了将该服务器部署到 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），仅供参考