使用GoHumanLoop提升模型上下文MCP服务人机协同能力

一. 前言

Hi，大家好，继续我们GoHumanLoop系列，前面介绍如何使用GoHumanLoop将Agent链接到第三方平台流程管理中，如邮箱、企业微信和飞书等，今天我们介绍一下如何在构建MCP服务时，通过GoHumanLoop提升MCP服务人机协同能力，针对高风险高决策的请求增加人类协同操作，确保整个Agent工作流按照人类预期开展。

这是 GoHumanLoop系列的第六篇文章，之前文章如下：

使用GoHumanLoop拓展AI Agent人机协同边界，这次连接到飞书

使用GoHumanLoop将你的AI Agent连接到企业微信（二）

使用GoHumanLoop将你的AI Agent连接到企业微信（一）

使用GoHumanLoop 在 CrewAI 高效构建Human-in-the-Loop流程

使用GoHumanLoop增强你的Agent控制 - LangGraph 中的实践

感兴趣的朋友可以阅读往期文章，进一步了解GoHumanLoop

本期我将给大家介绍MCP与GoHumanLoop如何搭配使用，我们上车吧～ 🚗

二. 当前MCP人机交互分析 - Elicitation

Elicitation

人机协同 — Elicitation Flow 是MCP 2025-06-18中新增的功能。模型上下文协议（MCP）为服务器在交互过程中通过客户端请求用户额外信息提供了一种标准化方式。这种流程允许客户端在保持对用户交互和数据共享控制的同时，使服务器能够动态收集必要的信息。服务器使用JSON模式从用户处请求结构化数据，以验证响应。

LLM

elicitationRequest

continueElicitation

此工作流程支持实际应用，例如：

交互式表单填写阐明用户意图收集增量数据确认不明确或部分输入

Elicitation 需要依赖AI应用实现人机交互，从用户获取信息并返回给MCP服务端，然后服务端继续执行。

为了信任、安全与保障：
服务器不得使用引诱方式请求敏感信息。
AI应用遵循的标准：
提供用户界面，明确显示哪个服务器正在请求信息允许用户在发送前查看和修改他们的回复尊重用户隐私，并提供明确的拒绝和取消选项

三. GoHumanLoop 和 Elicitation对别分析

GoHumanLoop：是一个Python库，使AI Agent能够在关键阶段动态请求人类输入（批准/反馈/对话）。

核心功能：

人类在环控制：让AI代理系统暂停并升级决策，增强安全性和信任。多渠道集成：支持终端、电子邮件、OA平台(企业微信、飞书等)和支持多个LangGraph/CrewAI等框架灵活的工作流程：结合自动推理和人工监督，确保人工智能操作的可靠性。

通过连接自主代理和人类判断，确保负责任的AI部署。

两者核心共同点

人机协同目标
两者均聚焦于构建"Human-in-the-loop"（HITL）系统，通过人类干预提升AI系统的安全性与可靠性。例如：

interrupt

安全控制机制

decline

cancel

create_resume_command

标准化交互范式

accept

decline

cancel

对比分析

维度	Elicitation（MCP协议）	GoHumanLoop（Python库）	共同目标
1. 协同目标	通过 JSON Schema 引导用户输入结构化数据（邮箱、年龄等）	通过 `interrupt` 机制强制暂停AI流程，等待人工介入（如 LangGraph）	构建闭环人机协作： • 人类补充关键信息或审核决策 • 确保AI行为可控、结果可靠
2. 安全控制	• 禁止请求敏感数据（内置规则） • 支持响应状态验证（decline/cancel）	• 强制人工审批所有工具调用 • 提供标准化恢复接口（`create_resume_command`）	双重防护敏感操作： • 入口拦截（禁止敏感请求） • 过程管控（人工审批关键动作）
3. 交互范式	定义三态响应模型： `accept`/`decline`/`cancel`	实现审批状态管理：支持跨平台适配（邮箱、企业微信、飞书）	标准化人机协议： • 明确交互状态机 • 兼容复杂流程（多轮、异步）

如GoHumanLoop可以灵活的针对各类方法、工具等提供异步的人机交互流程，不依赖于客户端APP的人机交互实现。（如上图所示），可以提高整个Agent功能在人机交互上面实现的效率，尤其是需要长时间运行的Agent和依赖审批的高危操作。

四. MCP服务使用GoHumanLoop

在MCP服务上使用GoHumanLoop很简单，只需要使用对应装饰器适配对应工具方法即可，我们采用fastmcp实现一个 MCP服务，如下代码所示

# /// script# requires-python = ">=3.10"# dependencies = [# "gohumanloop>=0.0.12",# "fastmcp>=2.6.0"]# ///from fastmcp import FastMCP, Clientfrom gohumanloop import DefaultHumanLoopManager, HumanloopAdapter, TerminalProvider# 设置环境变量# 创建 GoHumanLoopManager 实例manager = DefaultHumanLoopManager(    TerminalProvider(name="TerminalProvider"))adapter = HumanloopAdapter(    manager=manager,    default_timeout=600,  # 默认超时时间为10分钟)mcp = FastMCP(name="CalculatorServer")@mcp.tool()@adapter.require_approval()async def add(a: int, b: int,) -> int:    """Adds two integer numbers together."""    return a + b@mcp.tool()@adapter.require_approval()async def multiply(a: int, b: int) -> int:    """Multiply two numbers"""    return a * b@mcp.tool(exclude_args=["huamninfo"])@adapter.require_info(ret_key="huamninfo")async def feedback(add_result: int, huamninfo: dict={}) -> str:    """Gets feedback from human response.    Args:        add_result: The result from previous addition operation    Returns:        The human response string    """    print(f"Get Human Response: {huamninfo}")    info = huamninfo["response"]    return infoif __name__ == "__main__":    # This runs the server, defaulting to STDIO transport    mcp.run(transport="streamable-http", port=3000)

通过HumanloopAdapter实现的装饰器require_approval，封装到对应工具方法即可,如add和multiply，以上HumanloopAdapter的请求提供者TerminalProvider是指使用终端作为审批信息输入渠道。

我们再看一个例子

from mcp.server.fastmcp import FastMCPfrom gohumanloop import DefaultHumanLoopManager, HumanloopAdapter, APIProvider,  get_secret_from_envimport os#设置环境变量os.environ["GOHUMANLOOP_API_KEY"] = "46cf87c5-9d08-4027-b72a-f0a91f27298b"# 创建 GoHumanLoopManager 实例manager = DefaultHumanLoopManager(    APIProvider(        name="ApiProvider",        api_base_url="http://127.0.0.1:9800/api", # 换成自己飞书应用的URL        api_key=get_secret_from_env("GOHUMANLOOP_API_KEY"),        default_platform="feishu"    ))# 创建 LangGraphAdapter 实例adapter = HumanloopAdapter(    manager=manager,    default_timeout=600,  # 默认超时时间为10分钟)mcp = FastMCP("Weather")@mcp.tool()@adapter.require_info()async def get_weather(location: str, info_result:dict={}) -> str:    current_date = info_result["response"]    """Get weather for location."""    return f"It's always sunny in {location} on {current_date}"if __name__ == "__main__":    mcp.run(transport="streamable-http")

上面例子HumanloopAdapter采用了APIProvider来进行人机协同，可以将该MCP服务链接到飞书中，可以在飞书中进行操作的信息获取，进而帮助工具完成操作。如上述获取天气信息中，通过info_resuslt获取到用户从飞书中返回的当前日期信息

使用飞书需要对应 API Service，可以参考 github.com/ptonlix/goh… 进行适配使用

我们langchain使用一下以上工具

from langchain_mcp_adapters.client import MultiServerMCPClientfrom langgraph.prebuilt import create_react_agentclient = MultiServerMCPClient(    {        "math": {            "url": "http://localhost:3000/mcp/",            "transport": "streamable_http",        },        "weather": {            # Make sure you start your weather server on port 8000            "url": "http://localhost:8000/mcp/",            "transport": "streamable_http",        }    }<img src=")" alt="" width="50%" />async def main():    tools = await client.get_tools()    agent = create_react_agent("openai:deepseek-chat", tools)    math_response = await agent.ainvoke({"messages": "what's (3 + 5) x 12?"})    print(f"Question: what's (3 + 5) x 12? \nMath Response: {math_response['messages'][-1].content}\n")    weather_response = await agent.ainvoke({"messages": "what is the weather in nyc?"})    print(f"Question: what is the weather in nyc?\nWeather Response: {weather_response['messages'][-1].content}\n")if __name__ == "__main__":    import asyncio    asyncio.run(main())