自架代理程式作為 MCP 工具

Note

.NET 即將提供對自行託管 MCP 工具的支援。

Note

目前 Go 尚未支援自架 MCP 工具。

使用 agent-framework-hosting-mcp 將 Agent Framework 代理程式或工作流程在原生 Model Context Protocol SDK 中公開為工具。 該套件不會選擇網頁框架或包裝 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 新增應用程式擁有的 JSON Schema 屬性,這些屬性仍保留在原始 MCP 參數中,並 chat_option_parameters 新增可明確複製至 Agent Framework 聊天選項的屬性。

持續執行每通話一個會話

傳入現有的 AgentStatesession_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 get /run/set 這個會話序列;你的應用程式必須認證或授權會話識別碼,並序列化同一會話的並行呼叫,就像範例中對每個會話 asyncio.Lock所做的 。 這不是 previous_response_id-style 分支——需要分叉對話的應用程式應該接受不同的來源與目的 ID,複製來源會話,並將結果儲存在目的鍵下。

將工作流程當作工具來架設

WorkflowMCPTool 從工作流程的啟動-執行器輸入型態衍生出一個原生 MCP 工具,並將完成的工作流程輸出轉換。 Dataclass、Pydantic 及其他物件形狀輸入成為頂層 MCP 參數;原始輸入會被包裹在可配置的參數名稱中:

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

工作流程實例會保留執行狀態,因此,需要進行獨立呼叫的應用程式應如上所示提供搭配 cache_target=FalseWorkflowState 工廠。 檢查點還原、人機回應與持續識別碼仍由應用程式擁有;若工作流程要求外部輸入,介面卡會升高而非回傳空的成功結果。

關於完整的可執行伺服器集合——包括由裝飾函式衍生結構的 FastMCP 變體——請參見 MCP 主機範例

Important

將 MCP 會話識別碼及任何應用程式定義 session_id 的參數視為不可信輸入。 在使用前,先驗證並授權呼叫者,以載入或儲存會話狀態,並從已認證的租戶、使用者或工作區而非原始值中推導出持久的分割。

下一步

深入探討: