自承载代理作为 MCP 工具

注释

对在 .NET 中自承载 MCP 工具的支持即将推出。

注释

Go 目前尚不支持自托管 MCP 工具。

使用 agent-framework-hosting-mcp 在原生 模型上下文协议 SDK 上将 Agent Framework 代理或工作流公开为工具。 包不选择 Web 框架或包装 MCP SDK 服务器生命周期;应用程序仍拥有 Server、处理程序注册、传输、会话密钥策略、身份验证、授权和部署。

pip install --pre agent-framework-hosting-mcp

在协议边界处转换

mcp_to_run(...) 将验证的 MCP 工具参数转换为代理框架消息和所选聊天选项,并将 mcp_from_run(...) 已完成的响应转换为本机 MCP ContentBlock 值。 当应用程序的工具协定需要完全自定义的本机架构和处理程序时,请直接使用这两个函数:

@server.list_tools()
async def list_tools() -> list[types.Tool]:
    """Return the app-owned native MCP tool definition."""
    return [
        types.Tool(
            name="run_agent_manually",
            description=agent.description or "",
            inputSchema={
                "type": "object",
                "properties": {
                    TASK_ARGUMENT: {
                        "type": "string",
                        "description": "The request for the hosted agent.",
                    },
                    **CHAT_OPTION_ARGUMENTS,
                },
                "required": [TASK_ARGUMENT],
                "additionalProperties": False,
            },
        )
    ]


@server.call_tool()
async def call_tool(name: str, arguments: dict[str, object] | None) -> list[types.ContentBlock]:
    """Convert, run, and render without the agent-backed adapter."""
    if name != "run_agent_manually":
        raise ValueError(f"Unknown MCP tool: {name}")
    run = mcp_to_run(
        arguments,
        argument_name=TASK_ARGUMENT,
        chat_option_arguments=CHAT_OPTION_ARGUMENTS,
    )
    result = await agent.run(run["messages"], options=run["options"])
    return mcp_from_run(result)

仅将列出的 chat_option_arguments 参数名称复制到 run["options"];其他 MCP 参数在消息的原始表示形式上保持可用,但不会转发到模型客户端。

将代理作为一个已生成的工具进行托管

AgentMCPTool 从代理派生原生工具名称、描述和模式,并使列表、解析、执行和结果转换始终保持一致,从而避免二者出现偏离:

agent_tool = AgentMCPTool(
    agent,
    name="run_agent",
    argument_description="The request for the hosted agent.",
    chat_option_parameters={
        "reasoning_effort": {
            "type": "string",
            "enum": ["low", "medium", "high"],
            "description": "Optional reasoning effort for models that support it.",
        }
    },
)


@server.list_tools()
async def list_tools() -> list[types.Tool]:
    """Describe the app-owned MCP tool schema."""
    return await agent_tool.list_tools()


@server.call_tool()
async def call_tool(name: str, arguments: dict[str, object] | None) -> list[types.ContentBlock]:
    """Run the app-owned tool with native MCP and Agent Framework values."""
    return await agent_tool.call_tool(name, arguments)

AgentMCPTool 除非被覆盖,否则将使用该代理的名称和说明。 parameters 添加在原始 MCP 参数中保持可用的应用自有 JSON Schema 属性,而 chat_option_parameters 添加其值会被显式复制到 Agent Framework 聊天选项中的属性。

每次调用时保持会话

传入现有的 AgentState 和一个 session_id_parameter,使使用相同的、由应用定义的不透明 session_id 的重复调用能够继续同一对话:

session_locks: dict[str, asyncio.Lock] = {}


@server.list_tools()
async def list_tools() -> list[types.Tool]:
    """Return the agent-derived MCP tool definition."""
    return await agent_tool.list_tools()


@server.call_tool()
async def call_tool(name: str, arguments: dict[str, object] | None) -> list[types.ContentBlock]:
    """Serialize calls per app-owned session before using ``AgentState``."""
    session_id = arguments.get("session_id") if arguments else None
    if not isinstance(session_id, str) or not session_id:
        raise ValueError("MCP tool argument 'session_id' must be a non-empty string.")
    lock = session_locks.setdefault(session_id, asyncio.Lock())
    async with lock:
        return await agent_tool.call_tool(name, arguments)

AgentMCPTool 仅执行 AgentState 会话获取/运行/设置序列;应用程序必须对会话标识符进行身份验证或授权,并序列化同一会话的并发调用,就像示例对每个会话 asyncio.Lock执行的操作一样。 这不是 previous_response_id样式分支 — 需要分叉会话的应用程序应接受单独的源 ID 和目标 ID,复制源会话,并将结果存储在目标密钥下。

将工作流作为工具托管

WorkflowMCPTool 根据工作流启动执行器的输入类型生成一个原生 MCP 工具,并转换已完成工作流的输出。 数据类、Pydantic 和其他对象形输入成为顶级 MCP 参数;基元输入包装在可配置的参数名称中:

server = Server("agent-framework-hosting-mcp-workflow-sample")
workflow_tool = WorkflowMCPTool(
    WorkflowState(create_workflow, cache_target=False),
    name="draft_content",
)

工作流实例会保留执行状态,因此,需要独立调用的应用程序应提供 WorkflowState 工厂并配合 cache_target=False 使用,如上所示。 检查点还原、人工循环响应和延续标识符仍保留应用程序拥有;如果工作流请求外部输入,适配器将引发而不是返回空成功的工具结果。

有关完整的可运行服务器集(包括从修饰函数派生其架构的 FastMCP 变体),请参阅 MCP 托管示例

Important

将 MCP 会话标识符和任何应用定义 session_id 参数视为不受信任的输入。 在使用二者之一加载或保存会话状态之前,先对调用方进行身份验证和授权,并且应基于已验证的租户、用户或工作区来确定持久分区,而不是基于原始值。

后续步骤

更深入: