Agentes autohospedados como herramientas de MCP

Note

La compatibilidad con la herramienta MCP de autohospedaje en .NET estará disponible próximamente.

Note

La compatibilidad con herramientas MCP autoalojadas no está disponible actualmente para Go.

Use agent-framework-hosting-mcp para exponer un agente de Agent Framework o un flujo de trabajo como herramienta en el SDK nativo de Model Context Protocol. El paquete no elige un marco web ni encapsula el ciclo de vida del servidor SDK de MCP; tu aplicación sigue encargándose del Server, del registro de controladores, del transporte, de la directiva de claves de sesión, de la autenticación, de la autorización y del despliegue.

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

Conversión en el límite del protocolo

mcp_to_run(...) convierte los argumentos de la herramienta MCP validados en mensajes de Agent Framework y opciones de chat seleccionadas y mcp_from_run(...) convierte una respuesta completa en valores de MCP ContentBlock nativos. Use estas dos funciones directamente cuando el contrato de herramientas de una aplicación necesite un esquema y un controlador 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)

Solo se copian en run["options"] los nombres de argumentos enumerados en chat_option_arguments; los demás argumentos de MCP siguen estando disponibles en la representación sin procesar del mensaje, pero no se reenvían al cliente del modelo.

Hospedar un agente como una herramienta generada

AgentMCPTool obtiene de un agente el nombre, la descripción y el esquema de la herramienta nativa, y mantiene alineados el listado, el análisis sintáctico, la ejecución y la conversión de resultados para evitar que ambos se desincronicen:

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 el nombre y la descripción del agente a menos que se invalide. parameters agrega propiedades de esquema JSON propiedad de la aplicación que permanecen disponibles en los argumentos MCP sin procesar y chat_option_parameters agrega propiedades cuyos valores se copian explícitamente en las opciones de chat de Agent Framework.

Mantener una sesión por llamada

Pase un AgentState existente y un session_id_parameter para permitir que las llamadas repetidas con el mismo session_id, opaco y definido por la aplicación, mantengan la misma conversación:

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 solo realiza la secuencia de sesión AgentState get/run/set; su aplicación debe autenticar o autorizar el identificador de sesión y serializar las llamadas concurrentes para la misma sesión, como hace el ejemplo con un asyncio.Lock por sesión. Esto no es una bifurcación al estilo de previous_response_id: una aplicación que necesite bifurcar una conversación debe aceptar identificadores de origen y destino distintos, copiar la sesión de origen y almacenar el resultado bajo la clave de destino.

Hospedar un flujo de trabajo como herramienta

WorkflowMCPTool genera una herramienta MCP nativa a partir del tipo de entrada start-executor de un flujo de trabajo y convierte las salidas de los flujos de trabajo completados. Dataclass, Pydantic y otras entradas con forma de objeto se convierten en argumentos MCP de nivel superior; Las entradas primitivas se encapsulan en un nombre de argumento configurable:

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

Las instancias de flujo de trabajo conservan el estado de ejecución, por lo que las aplicaciones que necesitan invocaciones independientes deben proporcionar una factoría WorkflowState con cache_target=False, como se muestra anteriormente. La restauración de puntos de control, las respuestas con intervención humana y los identificadores de continuación siguen perteneciendo a la aplicación; si un flujo de trabajo solicita entrada externa, el adaptador lanza una excepción en lugar de devolver un resultado vacío y exitoso de la herramienta.

Para obtener el conjunto completo de servidores ejecutables , incluida la variante FastMCP que deriva su esquema de una función decorada, consulte los ejemplos de hospedaje de MCP.

Importante

Considere el identificador de sesión de MCP y cualquier argumento session_id definido por la aplicación como entradas no confiables. Autentique y autorice al emisor de la llamada antes de usar cualquiera de los dos para cargar o guardar el estado de la sesión, y obtenga la partición duradera a partir del inquilino, el usuario o el espacio de trabajo autenticados, en lugar del valor bruto.

Pasos siguientes

Más información sobre el hospedaje de A2A

Vaya más profundamente:

  • Last updated on