Agentes autohospedados como herramientas MCP

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

Convertir 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 chat_option_arguments los nombres de argumentos enumerados en run["options"]; 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.

Alojar 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 utiliza el nombre y la descripción del agente, a menos que se sobrescriban. 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 deriva una herramienta MCP nativa a partir del tipo de entrada start-executor de un flujo de trabajo y convierte las salidas del flujo de trabajo completado. 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 siendo propiedad de la aplicación; si un flujo de trabajo solicita una entrada externa, el adaptador genera un error en lugar de devolver un resultado vacío de la herramienta indicando que la operación se ha realizado con éxito.

Para ver el conjunto completo de servidores ejecutables —incluida la variante FastMCP, cuyo esquema se deriva de una función decorada—, consulta los ejemplos de hospedamiento de MCP.

Pasos siguientes

Vaya más profundamente:

• Información general sobre el autohospedaje
• Respuestas de OpenAI
• Telegrama