用 mcp-agent 构建 Reference Agent Server从 MCP 工具、Elicitation 到云端部署的完整实战【免费下载链接】mcp-agentBuild effective agents using Model Context Protocol and simple workflow patterns项目地址: https://gitcode.com/GitHub_Trending/mc/mcp-agent这篇技术指南以 mcp-agent 仓库中的 Reference Agent Server 示例为核心源码位于 src/mcp_agent/data/examples/mcp_agent_server/reference完整讲解如何把一个MCPApp变成一个可直接被任意 MCP 客户端连接的 SSE 服务端。你将掌握app.tool/app.async_tool工具声明、通过app.logger上报通知、向客户端反向代理 Elicitation 确认与 LLM Sampling以及注册 Prompts/Resources最后还能把同一套代码一键部署到云端。一、Reference Agent Server 是什么Reference Agent Server 是 mcp-agent 仓库中一个“干净、强类型”的示例服务端它的定位是展示把完整的 Agent 能力封装成 MCP 服务器后的标准写法。从 README.md 可以看到它集中演示了六类能力Agent 行为一个内部 Agent 组合了fetchfilesystem两个 MCP 服务器和一个 LLM对外表现为一个工具工具声明同时使用app.tool与app.async_tool两种装饰器通知与日志通过app.logger向上游客户端发送日志/通知Elicitation用户确认把需要用户确认的请求代理proxy给上游客户端SamplingLLM 调用使用简单的RequestParams触发一次 LLM 采样Prompts 与 Resources在 FastMCP 服务器上注册演示用的提示词和资源。这六个能力覆盖了「把 Agent 应用反向暴露为 MCP Server」的核心场景。整个示例只有三个文件server.py服务端、client.py最小测试客户端和README.md说明文档非常适合作为自定义 Agent Server 的起点模板。二、快速启动一行命令跑起 SSE 服务在示例目录src/mcp_agent/data/examples/mcp_agent_server/reference下执行uv run server.py服务随即在http://127.0.0.1:8000/sse启动一个基于 SSEServer-Sent Events的 MCP 服务端。README 同时提供了配套的最小客户端用于验证uv run client.py客户端通过 SSE 连接服务器、设置日志级别并依次调用四类能力对应的工具finder_tool—— Agent LLM MCP 服务器组合notify—— 日志/通知sample_haiku—— LLM 采样confirm_action—— Elicitation 确认提示。从实现看服务端在main()中通过create_mcp_server_for_app(agent_app)把整个MCPApp包装为 FastMCP 服务器再调用mcp_server.run_sse_async()启动 SSE 传输见 server.py。create_mcp_server_for_app定义在 src/mcp_agent/server/app_server.py它负责从应用配置读取授权设置auth_settings_config配置 FastMCP 的AuthSettings生成一个 app 专属的 lifespan在服务生命周期内完成app.initialize()、构建ServerContext在 lifespan 中自动注册工作流工具与app.tool/app.async_tool声明的函数工具。因此你只需要专注业务代码工具注册、上下文绑定、生命周期管理都由框架代劳。三、工具声明app.tool与app.async_tool示例中四个工具分别演示了两种声明方式其中finder_tool是最能体现 mcp-agent 特色的一个它本身就是一个封装在工具里的 Agent。3.1finder_tool工具内的 Agent 多 MCP 服务器app.tool(namefinder_tool) async def finder_tool(request: str, app_ctx: Optional[AppContext] None) - str: Agent that can use filesystemfetch and an LLM to answer the request. _app app_ctx.app if app_ctx else app ctx _app.context try: if filesystem in ctx.config.mcp.servers: ctx.config.mcp.servers[filesystem].args.extend([os.getcwd()]) except Exception: pass agent Agent( namefinder, instruction( Use MCP servers to fetch and read files, then answer the users query concisely. ), server_names[fetch, filesystem], contextctx, ) async with agent: llm await agent.attach_llm(OpenAIAugmentedLLM) return await llm.generate_str(messagerequest)关键点工具签名中的app_ctx这是一个由框架自动注入的AppContext即 mcp-agent 的Context见 server.py工具内部通过app_ctx.app拿到MCPApp实例。从 app.py 的实现看框架会在运行时用inspect.signature探测函数是否声明了名为app_ctx或注解为mcp_agent.core.context.Context的参数并在调用时自动注入当前工作流上下文动态扩展 MCP 服务器参数通过ctx.config.mcp.servers[filesystem].args.extend([os.getcwd()])在运行时把当前工作目录追加为 filesystem 服务器的挂载目录实现“工具被调用时按需挂载文件系统”Agent 组合Agent绑定fetch与filesystem两个服务器attach_llm(OpenAIAugmentedLLM)显式选择 OpenAI 增强型 LLM然后generate_str一次生成回答。3.2notify同步工具 结构化日志级别app.tool(namenotify) def notify( message: str, level: Literal[debug, info, warning, error] info, app_ctx: Optional[AppContext] None, ) - str: Send an upstream log/notification at the requested level. _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注意notify是同步函数使用的是app.tool而非app.async_tool。它把四种日志级别收敛为一个可被 MCP 客户端调用的工具并通过app.logger在 app.py 中按mcp_agent.{app_name}命名空间创建、且绑定了请求上下文把日志事件关联到当前上游会话实现“通知从服务器反向推送到客户端”。3.3app.tool与app.async_tool的底层机制从 src/mcp_agent/app.py 的实现可以看到app.tool装饰器并非简单地注册一个函数而是先调用validate_tool_schema(fn, tool_name)对函数做 JSON Schema 可转换性预校验通过_create_workflow_from_function把普通函数动态转换成Workflow子类AutoWorkflow_{name}并让该类的run负责在调用时注入app_ctx/ FastMCPContext把工具信息存入_declared_tools列表延迟到create_mcp_server_for_app()的 lifespan 阶段才真正注册为 FastMCP 工具见 app_server.py 中的create_declared_function_tools。app.async_tool的差异仅在于mark_sync_toolFalseapp.py表示该工具以异步工作流方式执行、可被run/get_status等标准工作流工具包装。这也解释了为什么示例中finder_tool、sample_haiku、confirm_action用 async 版本而notify用同步版本——前者可能耗时涉及 LLM 调用或等待用户输入后者即时返回。四、Elicitation把用户确认反向代理给上游客户端confirm_action演示了 mcp-agent 的 elicitation诱导式提问能力当 Agent 服务端需要用户确认某个操作时它可以把问题连同结构化的 JSON Schema 一起发给上游 MCP 客户端由客户端展示给最终用户。app.tool(nameconfirm_action) async def confirm_action( action: str, app_ctx: Optional[AppContext] None, ) - str: _app app_ctx.app if app_ctx else app upstream getattr(_app.context, upstream_session, None) class ConfirmBooking(BaseModel): confirm: bool Field(descriptionConfirm action?) notes: str Field(default, descriptionOptional notes) schema: ElicitRequestedSchema ConfirmBooking.model_json_schema() if upstream is not None: result await upstream.elicit( messagefDo you want to {action}?, requestedSchemaschema ) ...这段代码的关键设计优先上游转发若上下文存在upstream_session即当前服务端是被某个 MCP 客户端连接着的则直接调用upstream.elicit(...)把确认请求发送给客户端客户端一侧的响应处理逻辑位于 src/mcp_agent/mcp/mcp_agent_client_session.py同样优先向上游透传只有在上游不可用时才降级到本地elicitation_handler且任何KeyboardInterrupt/TimeoutError都会返回ElicitResult(actioncancel)保证流程可中断、不悬挂结构化确认结果确认请求携带ConfirmBooking.model_json_schema()客户端按 Schema 收集confirm布尔与notes可选备注字段。返回结果按action字段区分accept/accepted表示用户接受随后校验内容并回读备注decline表示拒绝本地降级回退当没有上游会话时工具回退到_app.context.elicitation_handler。仓库内置的console_elicitation_callback见 src/mcp_agent/elicitation/handler.py会在终端渲染蓝色面板并逐字段提示输入同时支持/decline、/cancel、/help三个斜杠命令兜底默认值两者都不可用时返回Action {action} confirmed by default避免工具因缺少确认渠道而抛错。在MCPApp构造时传入elicitation_callbackconsole_elicitation_callback即可启用本地兜底server.py。五、Sampling用简单 RequestParams 触发 LLM 调用sample_haiku演示了服务端主动发起一次 LLM 采样的最小写法——这对应 MCP 协议中的sampling/createMessage能力app.tool(namesample_haiku) async def sample_haiku(topic: str, app_ctx: Optional[AppContext] None) - str: _app app_ctx.app if app_ctx else app llm create_llm( agent_namesampling_demo, server_names[], instructionYou are a concise poet., context_app.context, ) req LLMRequestParams( maxTokens80, modelPreferencesModelPreferences(hints[]), systemPromptWrite a 3-line haiku., temperature0.7, use_historyFalse, max_iterations1, ) return await llm.generate_str(messagefHaiku about {topic}, request_paramsreq)要点拆解create_llm来自 src/mcp_agent/workflows/factory.py不依赖任何 MCP 服务器server_names[]是一个纯 LLM 实例RequestParams即LLMRequestParams来自 src/mcp_agent/workflows/llm/augmented_llm.py。示例中配置了maxTokens80、temperature0.7、use_historyFalse不维护多轮记忆、max_iterations1最多一次生成循环避免工具循环递归并显式传入systemPrompt约束输出格式ModelPreferences来自 src/mcp_agent/workflows/llm/llm_selector.py这里使用空hints[]表示交由模型选择器按默认策略决定具体模型。值得一提的是当服务端作为上游 MCP 客户端调用其它 MCP 服务器时收到的采样请求由 src/mcp_agent/mcp/sampling_handler.py 处理有上游会话则原样透传sampling/createMessage否则本地处理。本地处理会依次经历「人工审批请求 → 用 ModelSelector 选模型 → 生成 → 人工审批响应」四步任何一步被拒绝都会返回结构化ErrorData。六、Prompts 与 Resources注册 FastMCP 资源服务端除了工具外还在同一个 FastMCP 服务器上注册了演示用的资源和提示词server.pymcp_server.resource(demo://docs/readme)(_res_readme) mcp_server.resource(demo://{city}/weather)(_res_weather) mcp_server.prompt()(_prompt_echo)对应的资源与提示词清单类型标识说明Resourcedemo://docs/readme返回一段示例 README 内容Resourcedemo://{city}/weather动态参数资源返回简单的天气字符串Promptecho(message: str)返回Prompt: {message}其中demo://{city}/weather是带路径参数的模板资源city会在请求时被实际值替换。你可以使用任何支持列出 resources/prompts 的 MCP 客户端如 mcp-agent CLI 的mcp-agent chat或其它桌面客户端来探索这些内容——它们不是装饰性的Resources 可以为 LLM 提供上下文素材Prompts 则提供了可复用的提示词模板二者共同构成 Agent Server 对外暴露知识/模板的标准通道。七、最小客户端验证工具、日志与回调配套的 client.py 是一个自包含的 SSE 客户端直接演示了如何验证上面的服务settings Settings(execution_engineasyncio) app MCPApp( namereference_client, human_input_callbackconsole_input_callback, elicitation_callbackconsole_elicitation_callback, settingssettings, )强制 asyncio 执行引擎注释明确说明本地客户端侧需要 asyncio 才能承载 sampling/elicitation 回调Settings(execution_engineasyncio)内联声明服务器定义把reference_agent_server注册进server_registry指定transportsse、urlhttp://127.0.0.1:8000/sse然后通过gen_client(...)建立连接自定义会话工厂_make_session用MCPAgentClientSession包裹原生会话并传入logging_callbackon_server_log这样服务端app.logger发出的通知会被客户端以[SERVER LOG] [LEVEL] [logger] message的格式打印出来直观验证「通知反向推送」调用序列set_logging_level(info)→list_tools()列出工具 → 依次调用finder_tool请求“列出当前目录文件并总结”、notify、sample_haiku主题 clouds、confirm_action动作 proceed。运行uv run client.py时confirm_action会在客户端终端弹出确认提示你输入y或相应值后服务端工具会基于返回内容给出“Action proceed confirmed/cancelled”的结果从而完整体验完整的 elicitation 往返。八、配置与密钥管理服务端使用 mcp-agent 标准配置体系mcp_agent.secrets.yaml或环境变量存放 API 密钥如OPENAI_API_KEY。MCPApp初始化时会自动加载.env、.env.mcp-cloud等 dotenv 文件见 app.pymcp_agent.config.yaml声明mcp.servers如fetch、filesystem服务器及其启动参数与模型提供商的默认设置默认模型、temperature 等。finder_tool中通过ctx.config.mcp.servers读写配置正说明这一文件承载了服务器级配置。由于MCPApp(settings...)支持传入Settings实例或配置文件路径app.py你也可以在代码中覆盖配置。客户端侧同样可以内联构造Settings(execution_engineasyncio)而不依赖磁盘配置。九、可选一键部署到云端README 还给出了把同一份代码部署到 mcp-agent Cloud 的流程在mcp_agent.secrets.yaml中配置好 API 密钥在示例目录下执行uv run mcp-agent deploy reference-server部署完成后把返回的 URL 追加/sse作为 MCP 服务器地址填入你的 MCP 客户端如果部署环境要求鉴权则在客户端中以 bearer token 方式携带你的 mcp-agent API 密钥。这样本地 SSE 服务http://127.0.0.1:8000/sse与云端版本共用同一套代码与工具集从本地调试到生产发布的迁移成本几乎为零。注意云端部署是可选能力未配置云端账号时该步骤可跳过不影响本地体验。十、总结Reference Agent Server 把 mcp-agent 的核心能力浓缩在一个可运行的服务端 一个最小客户端中是理解「Agent 应用反向变成 MCP Server」的最佳入门样例工具层app.tool/app.async_tool自动把函数包装成工作流并延迟注册为 FastMCP 工具工具内部可以再嵌套完整的Agent组合多个 MCP 服务器与 LLM双向通道app.logger向上推送通知upstream.elicit/sampling/createMessage把用户确认与 LLM 采样请求代理给上游客户端且都带有本地降级回退协议面同时暴露 tools、prompts、resources并支持 SSE 传输任何标准 MCP 客户端均可直接使用工程化配置、密钥、日志、云端部署都有标准化的承接方式。如果你的目标是在 mcp-agent 之上快速搭建一个可复用的 Agent 服务端以 server.py 为骨架、以 client.py 为联调工具是最短路径。进一步的实现细节可继续研读 src/mcp_agent/server/app_server.py、src/mcp_agent/app.py 以及 src/mcp_agent/mcp/sampling_handler.py。【免费下载链接】mcp-agentBuild effective agents using Model Context Protocol and simple workflow patterns项目地址: https://gitcode.com/GitHub_Trending/mc/mcp-agent创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考