Agentes de auto-hospedagem como ferramentas MCP

Note

O suporte para autoalojamento de ferramentas MCP em .NET estará disponível em breve.

Note

O suporte para ferramentas MCP auto-hospedadas não está atualmente disponível para Go.

Utilize agent-framework-hosting-mcp para expor um agente ou fluxo de trabalho do Agent Framework como uma ferramenta no SDK nativo Model Context Protocol. O pacote não escolhe um framework web nem encapsula o ciclo de vida do servidor do SDK MCP; a sua aplicação continua a ser responsável por Server, pelo registo de manipuladores, pelo transporte, pela política de chaves de sessão, pela autenticação, pela autorização e pela implantação.

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

Converter no limite do protocolo

mcp_to_run(...) converte argumentos validados da ferramenta MCP em mensagens do Agent Framework e opções de chat selecionadas, e mcp_from_run(...) converte uma resposta completa em valores nativos do MCP ContentBlock . Use estas duas funções diretamente quando o contrato de ferramenta de uma aplicação necessita de um esquema e gestor nativos totalmente personalizados:

@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)

Apenas os nomes dos argumentos listados em chat_option_arguments são copiados para run["options"]; outros argumentos MCP permanecem disponíveis na representação bruta da mensagem, mas não são encaminhados para o cliente modelo.

Hospedar um agente como uma ferramenta gerada

AgentMCPTool obtém de um agente o nome nativo da ferramenta, a descrição e o esquema, e mantém alinhadas a listagem, a interpretação, a execução e a conversão de resultados, evitando que ambos se desalinhem:

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 usa o nome e a descrição do agente, salvo indicação em contrário. parameters adiciona propriedades do schema JSON pertencentes à aplicação que permanecem disponíveis nos argumentos MCP em bruto, e chat_option_parameters adiciona propriedades cujos valores são explicitamente copiados para as opções de chat do Agent Framework.

Manter sessão por chamada

Passe um AgentState existente e um session_id_parameter para permitir que chamadas repetidas com o mesmo session_id opaco definido pela aplicação continuem a mesma conversa:

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 apenas executa a sequência AgentState get/run/set da sessão; a sua aplicação tem de autenticar ou autorizar o identificador da sessão e serializar chamadas concorrentes para a mesma sessão, tal como o exemplo faz com um asyncio.Lock por sessão. Isto não é uma ramificação no estilo previous_response_id — uma aplicação que precise de bifurcar uma conversa deve aceitar IDs de origem e de destino separados, copiar a sessão de origem e armazenar o resultado sob a chave de destino.

Hospedar um fluxo de trabalho como ferramenta

WorkflowMCPTool gera uma ferramenta nativa de MCP a partir do tipo de entrada start-executor de um fluxo de trabalho e converte os resultados de saída de fluxos de trabalho concluídos. Dataclass, Pydantic e outras entradas em forma de objeto tornam-se argumentos MCP de topo; As entradas primitivas são envoltas num nome de argumento configurável:

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

As instâncias de fluxo de trabalho mantêm o estado de execução, pelo que as aplicações que necessitam de chamadas independentes devem fornecer uma fábrica WorkflowState com cache_target=False, como mostrado acima. A restauração de pontos de controlo, as respostas humanas no ciclo e os identificadores de continuação continuam a ser propriedade da aplicação; Se um fluxo de trabalho solicitar entrada externa, o adaptador gera em vez de devolver um resultado vazio e bem-sucedido da ferramenta.

Para o conjunto completo de servidores executáveis — incluindo a variante FastMCP que deriva o seu esquema a partir de uma função decorada — veja os exemplos de alojamento MCP.

Importante

Trate o identificador de sessão MCP e qualquer argumento definido session_id pela aplicação como entrada não confiável. Autentique e autorize o autor da chamada antes de utilizar qualquer um deles para carregar ou guardar o estado da sessão, e obtenha um particionamento persistente a partir do tenant, utilizador ou espaço de trabalho autenticados, em vez do valor em bruto.

Passos seguintes

Vai mais fundo: