Agentes hospedados en Foundry

Los agentes hospedados en Microsoft Foundry Agent Service permiten implementar aplicaciones de agente en contenedores en infraestructura administrada por Microsoft. La plataforma controla el escalado, la persistencia del estado de sesión, la seguridad y la administración del ciclo de vida para que pueda centrarse en la lógica del agente. Microsoft Foundry Hosted Agents ya está disponible con carácter general y admite agentes creados con tu propio código o con el marco de agentes que prefieras. En este artículo se describe específicamente la integración de hospedaje de Agent Framework.

Mediante la integración de alojamiento de Agent Framework, puede exponer un Agent a través del protocolo Foundry Responses o Invocations con código mínimo. Python también admite el hospedaje de un elemento nativo Workflow directamente, sin convertirlo en un agente.

Note

También puede implementar código de agente creado con otros marcos para agentes hospedados en Foundry mediante flujos de trabajo de la CLI de desarrollador (azd) de Azure. Para conocer los conceptos e instrucciones de implementación independientes del marco, consulte ¿Qué son los agentes hospedados? El resto de este artículo se centra en la integración de Agent Framework.

Cuándo usar agentes hospedados

Elija Agentes hospedados de Foundry cuando desee:

  • Infraestructura administrada : no es necesario configurar contenedores, servidores web ni reglas de escalado usted mismo.
  • Administración de sesiones integrada : la plataforma conserva y carga archivos en turnos y períodos $HOME de inactividad.
  • Identidad del agente dedicado — cada agente implementado obtiene su propia identidad Entra para el acceso seguro a modelos, herramientas y servicios posteriores.
  • Puntos de conexión compatibles con OpenAI : los clientes pueden interactuar con el agente mediante cualquier SDK compatible con OpenAI mediante el protocolo De respuestas.
  • Para los agentes de audio en tiempo real, use agentes alojados con Azure Speech en Foundry Tools (Voice Live) para la detección de actividad de voz en el servidor, la cancelación de eco y la reducción de ruido. Para obtener más información, consulte Uso de Voice Live con agentes hospedados.

Note

La integración de Python agent-framework-foundry-hosting se encuentra en versión preliminar. Agentes hospedados de Microsoft Foundry, el servicio de alojamiento administrado, ya está disponible de forma general.

Prerequisites

Para las pruebas locales, también necesita lo siguiente:

Instale el paquete NuGet de hospedaje:

dotnet add package Microsoft.Agents.AI.Foundry.Hosting --prerelease
  • Python 3.10 o posterior

Instale el paquete de hospedaje de versión preliminar, el cliente de Foundry y el paquete de autenticación de Azure:

pip install --pre agent-framework-foundry agent-framework-foundry-hosting azure-identity

En Foundry, la plataforma proporciona el contexto de usuario del autor de la llamada y el contexto de llamada; la infraestructura de hospedaje las usa para aislar el estado por usuario y reenviar el contexto de solicitud a los servicios Foundry. Las ejecuciones locales no reciben ese contexto de plataforma, por lo que las aplicaciones deben proporcionar sus propios controles de identidad y estado cuando sea necesario.

Protocolo de respuestas

El protocolo De respuestas es el punto de partida recomendado para la mayoría de los agentes. Expone un punto de conexión compatible con /responses OpenAI y la plataforma administra automáticamente el historial de conversaciones, el streaming y el ciclo de vida de la sesión.

En los agentes de Python hospedados, una respuesta que termina antes de tiempo tiene el estado incomplete. Los clientes de streaming reciben un evento de terminal response.incomplete , mientras que los clientes que no son de streaming reciben status establecido en incomplete. Un motivo de finalización content_filter se corresponde con incomplete_details.reason establecido en content_filter, y length se corresponde con max_output_tokens. Cualquier salida generada o contenido de rechazo sigue estando disponible en la respuesta.

using Azure.AI.AgentServer.Core;
using Azure.AI.Projects;
using Azure.Identity;
using Microsoft.Agents.AI;
using Microsoft.Agents.AI.Foundry.Hosting;

var projectEndpoint = new Uri(Environment.GetEnvironmentVariable("FOUNDRY_PROJECT_ENDPOINT")
    ?? throw new InvalidOperationException("FOUNDRY_PROJECT_ENDPOINT is not set."));
var deployment = Environment.GetEnvironmentVariable("FOUNDRY_MODEL")
    ?? Environment.GetEnvironmentVariable("AZURE_AI_MODEL_DEPLOYMENT_NAME")
    ?? "gpt-4o";

AIAgent agent = new AIProjectClient(projectEndpoint, new DefaultAzureCredential())
    .AsAIAgent(
        model: deployment,
        instructions: "You are a helpful AI assistant.",
        name: "my-agent");

var builder = AgentHost.CreateBuilder(args);
builder.Services.AddFoundryResponses(agent);
builder.RegisterProtocol("responses", endpoints => endpoints.MapFoundryResponses());

var app = builder.Build();
app.Run();

AgentHost.CreateBuilder crea un host de aplicación preconfigurado para el entorno de hospedaje de Foundry. AddFoundryResponses registra el agente con el controlador de protocolo de respuestas y MapFoundryResponses asigna el /responses punto de conexión HTTP.

import os

from agent_framework import Agent
from agent_framework.foundry import FoundryChatClient
from agent_framework_foundry_hosting import ResponsesHostServer
from azure.identity import DefaultAzureCredential

client = FoundryChatClient(
    project_endpoint=os.environ["FOUNDRY_PROJECT_ENDPOINT"],
    model=os.environ.get("FOUNDRY_MODEL") or os.environ["AZURE_AI_MODEL_DEPLOYMENT_NAME"],
    credential=DefaultAzureCredential(),
)

agent = Agent(
    client=client,
    instructions="You are a helpful AI assistant.",
)

server = ResponsesHostServer(agent)
server.run()

ResponsesHostServer envuelve tu agente y lo expone a través del protocolo Foundry Responses. El campo store del solicitante controla si se guardan la respuesta externa, la sesión administrada por el host y el estado de autorización. La history_source configuración selecciona de forma independiente quién proporciona el historial de modelos:

history_source Comportamiento del historial de modelos
"agent_server" (valor predeterminado) El anfitrión reconstruye la transcripción externa almacenada de Responses y deshabilita el almacenamiento del servicio downstream para evitar duplicar el historial.
"service" El host envía únicamente la entrada actual y guarda de forma privada el ID de continuación del servicio de almacenamiento. Una conversación de proveedor almacenada no puede bifurcarse de una respuesta anterior.
"agent" El host envía solo la entrada actual. Los valores predeterminados de almacenamiento del agente HistoryProvider o del servicio posterior gestionan el historial.

No combines "agent_server" ni "service" con un HistoryProvider habilitado para carga. El modo predeterminado también rechaza opciones fijas de continuación descendente, como conversation_id, previous_response_id y conversation. Use history_source="agent" para una implementación personalizada de SupportsAgentRun.

El parámetro del compilador response_store selecciona el backend para la persistencia de las respuestas externas. El parámetro anterior del constructor store es un alias obsoleto de response_store; ninguno de los parámetros establece el campo store del llamador para cada solicitud. Una solicitud con store=false es de un solo uso: no guarda el estado gestionado por el host, deshabilita el almacenamiento posterior compatible y no puede usar background=true.

El host posee el agente proporcionado y podría agregar proveedores de contexto específicos del hospedaje. No reutilice el agente con otro host ni invoquelo directamente después de la construcción del host.

El entorno de Responses conserva las llamadas nativas del equipo, las capturas de pantalla y las comprobaciones de seguridad. La aplicación debe ejecutar las acciones solicitadas y confirmar explícitamente las comprobaciones de seguridad. Para obtener el flujo completo, consulte Uso del equipo nativo.

Seleccione una instancia de agente o una fábrica

Tanto ResponsesHostServer como InvocationsHostServer aceptan, a través del parámetro agent, ya sea una instancia de agente o una función invocable sin argumentos, síncrona o asíncrona. El host reutiliza una instancia durante su vigencia. Una función invocable se ejecuta una vez por solicitud, y el agente devuelto pertenece a esa solicitud.

Utilice un callable cuando un agente normal conserve el estado mutable fuera de AgentSession. Los hosts conservan solo los almacenes de sesión, puntos de control y aprobación de funciones que admiten, no campos arbitrarios en un agente con ámbito de solicitud.

Warning

El hospedaje de un WorkflowAgent, como workflow.as_agent(), a través de agent= está en desuso. Un WorkflowAgent mantiene el estado del flujo de trabajo en la memoria entre ejecuciones, por lo que una instancia nunca debe atender solicitudes de distintos usuarios o conversaciones. Hospede el flujo de trabajo de forma nativa a través de workflow= y una fábrica consciente de solicitudes.

Hasta que migre un host de respuestas, una factoría que crea un flujo de trabajo nuevo, ejecutores, agentes encapsulados, clientes, proveedores y herramientas para cada solicitud es la forma heredada segura. Mantenga estable el nombre del flujo de trabajo y los identificadores del ejecutor para que el host pueda restaurar puntos de control. En el caso de un host de invocaciones, esta forma de factoría heredada solo se adapta a flujos de trabajo sin estado y de un solo turno porque el hospedaje del agente no restaura los puntos de control de flujo de trabajo.

Utilice también una fábrica cuando una integración transporte la identidad de la solicitud o posea recursos específicos de la solicitud. Por ejemplo, crea conexiones MCP, cajas de herramientas, proveedores de habilidades, clientes de búsqueda, proveedores de memoria y sus credenciales dentro de la fábrica cuando utilicen la llamada de la plataforma actual o el contexto de usuario. La reutilización de una conexión MCP para todo el proceso puede conservar la identidad de la solicitud que la abrió.

El host inicia y finaliza los agentes creados por la fábrica para cada solicitud. Agent administra clientes administrados por contexto y herramientas de MCP, pero la factoría debe cerrar cualquier otro proveedor, transporte o credencial que cree. No cierre los objetos compartidos que la aplicación proporcionó desde fuera de la fábrica.

Aloja un flujo de trabajo nativo con respuestas

Python puede hospedar un flujo de trabajo compilado directamente a través de workflow=. Un flujo de trabajo nativo requiere un parse_response callback que asigna la solicitud actual de Responses a una entrada de inicio tipada o al lote completo de respuestas pendientes:

from pydantic import BaseModel

from agent_framework_foundry_hosting import (
    CheckpointStoreProvider,
    HostedResponseRequest,
    ResponsesHostServer,
    WorkflowTurn,
)


class Ticket(BaseModel):
    text: str


def build_workflow(request: HostedResponseRequest):
    return build_fresh_workflow()


async def parse_response(request: HostedResponseRequest) -> WorkflowTurn[Ticket]:
    items = await request.get_input_items()
    if any(item.get("type") in ("function_call_output", "mcp_approval_response") for item in items):
        return WorkflowTurn(responses=await request.get_workflow_responses())

    text = await request.get_input_text()
    return WorkflowTurn(input=Ticket.model_validate_json(text or ""))


server = ResponsesHostServer(
    workflow=build_workflow,
    parse_response=parse_response,
    checkpoint_store_provider=CheckpointStoreProvider(
        allowed_checkpoint_types=[f"{Ticket.__module__}:{Ticket.__qualname__}"],
    ),
)

Utiliza una fábrica sincrónica o asincrónica sensible a las solicitudes para flujos de trabajo que puedan pausar, continuar o recuperar el trabajo en segundo plano. La fábrica debe devolver un grafo recién compilado con ejecutores, agentes, clientes, proveedores y herramientas mutables nuevos. Mantenga estable el nombre del flujo de trabajo y los identificadores del ejecutor para que el host pueda restaurar el punto de control exacto asociado a la respuesta externa.

El usuario de la plataforma de confianza y el entorno aislado de Foundry aíslan el estado del flujo de trabajo nativo. El host valida un lote de respuesta pendiente completo antes de consumir cualquier autorización de respuesta. Las respuestas obsoletas, parciales, duplicadas, reproducidas, entre usuarios y entre entornos de pruebas fallan antes de la ejecución del flujo de trabajo. Una solicitud con store=false no guarda el estado del flujo de trabajo y no puede devolver una pausa reanudable.

Para un flujo de trabajo heredado que acepta list[Message], use response_input_messages(request) para convertir solo el turno actual de Responses. No carga el historial externo previo ni descodifica las respuestas pendientes del flujo de trabajo. Hosting agent=workflow.as_agent() permanece disponible durante la versión beta actual, pero emite una advertencia de desuso. Para obtener ejemplos completos, consulte los ejemplos de flujo de trabajo de respuestas nativas.

Conservar el estado y controlar las conversaciones de larga duración

ResponsesHostServer y InvocationsHostServer configuran los almacenes de sesión persistentes de forma predeterminada. AgentSessionStoreProvider proporciona un FoundryAgentSessionStore; las sesiones de respuestas usan el almacén lógico agent_sessions, mientras que las sesiones de invocaciones usan el almacén invocation_sessions independiente. Estos almacenes usan Foundry State Store cuando se alojan y el almacenamiento basado en archivos del SDK cuando se ejecutan localmente.

En el caso de los agentes del flujo de trabajo de Responses, CheckpointStoreProvider proporciona un FoundryCheckpointStore. Los flujos de trabajo nativos de Respuestas e Invocaciones utilizan el mismo proveedor para sus puntos de control de continuación exactos. FunctionApprovalStoreProvider proporciona un FoundryFunctionApprovalStore para las aprobaciones pendientes de herramientas de agente. En su lugar, las respuestas a solicitudes y aprobaciones del flujo de trabajo nativas están enlazadas a los puntos de control del flujo de trabajo.

Cuando se ejecuta en Foundry, la instalación predeterminada de Python almacena el estado del espacio de nombres según el identificador de usuario de la plataforma y el identificador de sesión del sandbox de Foundry. También requieren un identificador de llamada a la plataforma para cada operación de estado. El identificador de llamada autoriza y correlaciona la operación; no es un identificador de conversación y no forma parte de la clave de almacenamiento.

En el caso de las respuestas, el FOUNDRY_AGENT_SESSION_ID configurado por la plataforma identifica el espacio aislado, y se rechaza cualquier agent_session_id diferente proporcionado por quien realiza la llamada. Para Invocations, el host verifica el parámetro de consulta enrutado agent_session_id con respecto al contexto de la solicitud. Si FOUNDRY_AGENT_SESSION_ID no está configurado, el parámetro de consulta debe estar presente, no vacío y coincidir con el contexto de solicitud. Los valores que faltan, duplicados o en conflicto se rechazan en lugar de usar un identificador de reserva del SDK.

Estas garantías se aplican a los almacenes hospedados predeterminados. Los proveedores de tiendas personalizadas deben implementar un aislamiento equivalente para el usuario y el entorno de pruebas, mantener el AgentSession.session_id interno separado de las claves de búsqueda del host y utilizar escrituras condicionales para que las solicitudes obsoletas no puedan sobrescribir las instantáneas más recientes. Las nuevas claves deben usar operaciones de escritura de solo creación en lugar de operaciones de actualización o inserción incondicionales. Consulte el ejemplo de almacenamiento personalizado para una implementación de Cosmos DB con escrituras y eliminaciones protegidas con ETag.

Con history_source="agent", el almacén de sesión configurado conserva el estado del proveedor llevado por AgentSession, incluidos los mensajes de InMemoryHistoryProvider.

Ambos hosts aceptan valores comprendidos entre StoreProvider[SessionStore] y agent_session_store_provider. El estado de sesión debe admitir la AgentSession serialización. Registre códecs para tipos de estado personalizados con register_state_type(); el estado restaurado no conserva la identidad de los objetos de Python. Los nuevos almacenes predeterminados caducan las sesiones 30 días después de su última escritura. Los proveedores personalizados controlan su propia retención.

Los almacenes predeterminados con ámbito no leen datos heredados sin ámbito agent_sessions, invocation_sessions, de punto de control o de aprobación de funciones. Inicie una nueva conversación de Responses en lugar de reutilizar un previous_response_id antiguo o un ID de conversación antiguo. Invocations comienza con una sesión vacía de Agent Framework en el almacén con ámbito.

Los registros cargados AgentSession usan condiciones de ETag. Si otra solicitud hace avanzar primero esa misma sesión, la escritura obsoleta falla en lugar de sobrescribir el estado más reciente. Esta comprobación no proporciona transacciones ni ejecución exactamente una vez para los efectos secundarios del agente o de la herramienta, por lo que las aplicaciones deben seguir coordinando las solicitudes que se solapan.

Para el almacenamiento específico de Responses, pase un StoreProvider a function_approval_store_provider o un ContextScopedStoreProvider a checkpoint_store_provider.

El trabajo en segundo plano externo utiliza el response.id visible para el llamante para realizar sondeos. El valor predeterminado background_source="agent_server" mantiene la ejecución en segundo plano en el host. Establece background_source="provider" únicamente con history_source="service" y un cliente de respuestas de almacenamiento que permita la reanudación. Si ResponsesServerOptions(resilient_background=True) también se establece, el host solo puede recuperar el sondeo del proveedor después de guardar el token de continuación privado. Haz que los efectos secundarios de las herramientas locales sean idempotentes, ya que un fallo antes de que se guarde el siguiente token puede reproducirlos.

Importe ResponsesServerOptions desde azure.ai.agentserver.responses, y páselo a ResponsesHostServer mediante el parámetro options. Las opciones de conversación de larga duración disponibles dependen del tipo de agente:

Capacidad Tipo de agente Requisitos y comportamiento
Recuperación en segundo plano del punto de control del flujo de trabajo Solo flujo de trabajo Establezca ResponsesServerOptions(resilient_background=True). Envíe la solicitud de respuestas con store=true y background=true. Después de un reinicio, el host reanuda el punto de control de flujo de trabajo duradero más reciente o reproduce la entrada original si no existe ningún punto de control. No configure el almacenamiento de puntos de control en el flujo de trabajo porque el host lo administra. Haga que los efectos secundarios externos sean idempotentes, porque el trabajo realizado después del último punto de control persistente podría repetirse.
Respuestas en segundo plano nativas del proveedor Agent que no forma parte del flujo de trabajo con un cliente de respuestas de almacenamiento Establezca history_source="service" y background_source="provider". Establece resilient_background=True cuando los tokens de continuación del proveedor guardados deban sobrevivir a un reinicio del host.
Conversaciones orientables No disponible temporalmente No establezca steerable_conversations=True. El host genera RuntimeError durante la construcción hasta que el SDK del servidor del agente administre de forma segura los giros de dirección rechazados.

Para ver implementaciones completas, consulte los ejemplos de almacenamiento personalizado, historial básico de Responses y procesamiento en segundo plano y flujo de trabajo resiliente de larga duración.

Leer archivos del entorno aislado alojado

Trata el $HOME persistente de un entorno de pruebas hospedado como un recurso enrutado por solicitud, no como un límite general del sistema de archivos. Acepte solo los archivos que la aplicación cargue explícitamente en un directorio dedicado, valide la identidad del espacio aislado actual y rechace rutas de acceso absolutas, recorridos, vínculos, archivos noregulares y contenido sobredimensionado o no válido.

Para el protocolo Responses, dirija una solicitud a una sesión alojada mediante el campo agent_session_id body. El selector de cadena de consulta se usa para invocaciones. Las cargas de sesión y los archivos del intérprete de código de Toolbox son recursos independientes; un archivo de entorno de pruebas cargado no se monta automáticamente en un contenedor de Toolbox. Consulte el ejemplo de archivos de sesión para las lecturas UTF-8 delimitadas y la guía para cargas locales y alojadas.

Opciones de solicitud de control

El host asigna campos nativos de generación de respuestas a las opciones de ejecución de Agent Framework. Por ejemplo, max_output_tokens se convierte en max_tokens y parallel_tool_calls se convierte en allow_multiple_tool_calls. Los valores aplanados de extra_body sobrescriben los valores nativos traducidos.

Utiliza el gancho prepare_options(request, options) síncrono o asíncrono para eliminar o sustituir las opciones del modelo del llamante antes de que se ejecute un agente regular. El gancho no puede establecer campos de identidad, almacenamiento, continuación o transporte controlados por el host. Para una implementación personalizada SupportsAgentRun que no puede aceptar las opciones del modelo en tiempo de ejecución, establezca unsupported_options en "warn" (valor predeterminado), "ignore"o "error".

Cuando una herramienta de MCP hospedada en Foundry requiere el consentimiento del usuario, ResponsesHostServer devuelve una respuesta incompleta con un oauth_consent_request elemento de salida. Presente su consent_link al usuario y, a continuación, continúe con el ID de la respuesta incompleta como previous_response_id después de que el usuario otorgue su consentimiento. El host conserva la sesión del agente para este reintento y expone solo los vínculos de consentimiento HTTPS absolutos.

Si el host conoce los orígenes de autorización esperados, restrinja los vínculos de consentimiento con allowed_oauth_consent_origins:

server = ResponsesHostServer(
    agent,
    allowed_oauth_consent_origins=[
        "https://logic-region.consent.azure-apihub.net",
        "https://auth.partner.example",
    ],
)

Si se omite la lista de permitidos, se mantiene la validación absoluta-HTTPS sin restringir el origen de destino. Proporcionar una lista vacía rechaza todos los vínculos de consentimiento. Configurar solo orígenes HTTPS exactos; se rechazan las entradas con una ruta de acceso, consulta o fragmento.

Protocolo de invocaciones

El protocolo Invocaciones proporciona control total sobre la solicitud y la respuesta HTTP. Úselo cuando necesite cargas personalizadas, procesamiento no conversacional o protocolos de streaming que no sean compatibles con OpenAI.

Con el protocolo de Invocaciones en C#, se implementa una implementación personalizada de InvocationHandler para procesar las solicitudes entrantes.

using Azure.AI.AgentServer.Core;
using Azure.AI.AgentServer.Invocations;
using Microsoft.Agents.AI;

var builder = AgentHost.CreateBuilder(args);

builder.Services.AddSingleton<AIAgent, MyAgent>();
builder.Services.AddInvocationsServer();
builder.Services.AddScoped<InvocationHandler, MyInvocationHandler>();

builder.RegisterProtocol("invocations", endpoints => endpoints.MapInvocationsServer());

var app = builder.Build();
app.Run();

El AddInvocationsServer método registra los servicios del protocolo de invocaciones. Implemente InvocationHandler para definir cómo procesa el agente cada solicitud.

Para una configuración ligera, use InvocationsHostServer del paquete agent_framework_foundry_hosting. Envuelve tu agente de forma similar a ResponsesHostServer y administra automáticamente las sesiones:

import os

from agent_framework import Agent
from agent_framework.foundry import FoundryChatClient
from agent_framework_foundry_hosting import InvocationsHostServer
from azure.identity import DefaultAzureCredential

client = FoundryChatClient(
    project_endpoint=os.environ["FOUNDRY_PROJECT_ENDPOINT"],
    model=os.environ.get("FOUNDRY_MODEL") or os.environ["AZURE_AI_MODEL_DEPLOYMENT_NAME"],
    credential=DefaultAzureCredential(),
)

agent = Agent(
    client=client,
    instructions="You are a friendly assistant. Keep your answers brief.",
    default_options={"store": False},
)

server = InvocationsHostServer(agent)
server.run()

InvocationsHostServer acepta los mismos formularios de fábrica de instancia o de solicitud descritos para el host de respuestas. Restaura las sesiones serializadas desde el almacén configurado, por lo que las conversaciones completadas pueden continuar después de reiniciar el host. Para conocer el comportamiento del almacenamiento, la retención y la personalización, consulta Persistir el estado y administrar conversaciones de larga duración.

Cuando se hospeda, Invocations utiliza el ámbito de solicitud verificado descrito en Persistir el estado y administrar conversaciones de larga duración. Trate AgentSession.session_id como un valor opaco; no analice ni dependa de su representación interna. Las ejecuciones locales mantienen su comportamiento de almacenamiento de usuario único existente.

Alojar un flujo de trabajo nativo con invocaciones

Pasa workflow= y una llamada de retorno parse_request explícita para hospedar un flujo de trabajo nativo. El callback posee el esquema JSON de la aplicación y devuelve un WorkflowTurn con una entrada tipada o el lote completo de respuestas pendientes:

from pydantic import BaseModel
from starlette.requests import Request

from agent_framework_foundry_hosting import (
    CheckpointStoreProvider,
    InvocationsHostServer,
    WorkflowTurn,
)


class Ticket(BaseModel):
    ticket_id: str
    question: str


class TicketDecision(BaseModel):
    approved: bool


def build_workflow(_request: Request):
    return build_fresh_workflow()


async def parse_request(request: Request) -> WorkflowTurn[Ticket]:
    payload = await request.json()
    stream = payload.get("stream", False)

    if "responses" in payload:
        decisions = {
            request_id: TicketDecision.model_validate(value)
            for request_id, value in payload["responses"].items()
        }
        return WorkflowTurn(responses=decisions, stream=stream)

    ticket = Ticket.model_validate(payload)
    return WorkflowTurn(input=ticket, stream=stream)


server = InvocationsHostServer(
    workflow=build_workflow,
    parse_request=parse_request,
    checkpoint_store_provider=CheckpointStoreProvider(
        allowed_checkpoint_types=[
            f"{Ticket.__module__}:{Ticket.__qualname__}",
            f"{TicketDecision.__module__}:{TicketDecision.__qualname__}",
        ],
    ),
)

Incluye todos los tipos de aplicaciones personalizadas que el flujo de trabajo guarde en la lista allowed_checkpoint_types del proveedor de puntos de control.

Los flujos de trabajo alojados requieren una factoría adaptada a la solicitud que devuelva un grafo construido nuevo con identificadores estables de flujo de trabajo y ejecutor. Un flujo de trabajo compilado directamente solo está disponible para una ejecución local única que no se detiene.

Las respuestas de flujo de trabajo que no son de streaming usan JSON de aplicación con una lista de eventos output. El streaming emite eventos enmarcados output y request_info, y emite done solo después de que se haya guardado el cursor exacto del flujo de trabajo. Trate el resultado transmitido como provisional hasta done. Los flujos de trabajo nativos no admiten legacy_wire_format=True.

El anfitrión valida las respuestas con respecto al punto de control pendiente exacto en el ámbito del usuario de confianza y del entorno de pruebas. Si un flujo de trabajo tiene varias solicitudes pendientes, responda al lote completo en una sola intervención. Para obtener un analizador ejecutable, un flujo de trabajo de tickets tipados, una lista de tipos de puntos de control permitidos y ejemplos de JSON/SSE, consulta el ejemplo de flujo de trabajo de invocaciones nativas.

Personalizar las solicitudes y respuestas de invocaciones

De forma predeterminada, POST /invocations acepta un objeto JSON con una cadena message, un objeto opcional options y un valor booleano stream opcional. Para aceptar una carga específica de la aplicación, pase una devolución de llamada sincrónica o asincrónica parse_request que devuelva InvocationRun(messages, options, stream). Usa prepare_options para filtrar o reemplazar una copia de las opciones de generación del llamador antes de que se ejecute el agente.

El host valida la salida del hook y rechaza los controles de identidad de la plataforma, almacenamiento, continuación y ejecución del agente. En el caso de los agentes que no aceptan opciones en tiempo de ejecución, establezca unsupported_options en "warn" (valor predeterminado), "ignore"o "error". Consulte el ejemplo de analizador de invocaciones para obtener una implementación completa.

Si la operación sin streaming se realiza correctamente, devuelve un JSON con el formato {"response": "..."}. El streaming usa eventos enviados por el servidor: una o varias tramas event: delta, seguidas de event: done en caso de éxito o de event: error en caso de fallo. Un flujo puede emitir deltas antes de que se produzca un error, por lo que los clientes deben considerar done, no un delta, como una finalización correcta. El host emite done solo después de finalizar la secuencia de respuesta y conserva .AgentSession Su session_id es el identificador de ruta del entorno aislado de la plataforma, no el AgentSession.session_id serializado.

Configure legacy_wire_format=True solo durante la migración de clientes existentes que requieran la respuesta anterior en texto sin formato y el flujo sin procesar de fragmentos de texto. Este modo de compatibilidad está en desuso y no convierte los errores en texto correcto. El host serializa las solicitudes de la misma sesión solo dentro de un proceso; Un conflicto de comparación y intercambio entre procesos puede seguir apareciendo después de los efectos de las herramientas externas.

El protocolo Invocaciones no reanuda las ejecuciones de flujo de trabajo que están pendientes o interrumpidas. Use el patrón de controlador personalizado en la sección siguiente cuando necesite un comportamiento de continuación de flujo de trabajo diferente.

Para un control total sobre el control de solicitudes, use InvocationAgentServerHost directamente desde el azure.ai.agentserver.invocations paquete e implemente su propio controlador de invocación:

import os
from collections.abc import AsyncGenerator

from agent_framework import Agent, AgentSession
from agent_framework.foundry import FoundryChatClient
from azure.ai.agentserver.invocations import InvocationAgentServerHost
from azure.identity import DefaultAzureCredential
from starlette.requests import Request
from starlette.responses import JSONResponse, Response, StreamingResponse

_sessions: dict[str, AgentSession] = {}

client = FoundryChatClient(
    project_endpoint=os.environ["FOUNDRY_PROJECT_ENDPOINT"],
    model=os.environ.get("FOUNDRY_MODEL") or os.environ["AZURE_AI_MODEL_DEPLOYMENT_NAME"],
    credential=DefaultAzureCredential(),
)

agent = Agent(
    client=client,
    instructions="You are a friendly assistant. Keep your answers brief.",
    default_options={"store": False},
)

app = InvocationAgentServerHost()


@app.invoke_handler
async def handle_invoke(request: Request):
    """Handle streaming multi-turn chat."""
    data = await request.json()
    session_id = request.state.session_id
    stream = data.get("stream", False)
    user_message = data.get("message", None)

    if user_message is None:
        return Response(content="Missing 'message' in request", status_code=400)

    session = _sessions.setdefault(session_id, AgentSession(session_id=session_id))

    if stream:

        async def stream_response() -> AsyncGenerator[str]:
            async for update in agent.run(user_message, session=session, stream=True):
                yield update.text

        return StreamingResponse(
            stream_response(),
            media_type="text/event-stream",
            headers={"Cache-Control": "no-cache", "Connection": "keep-alive"},
        )

    response = await agent.run([user_message], session=session, stream=stream)
    return JSONResponse({"response": response.text})


if __name__ == "__main__":
    app.run()

Warning

El almacén de sesión en memoria del ejemplo del controlador personalizado se pierde al reiniciar. Use almacenamiento duradero (por ejemplo, Cosmos DB) en producción.

Para obtener una implementación completa de invocaciones, consulte el ejemplo de Telegram hospedado por Foundry. Coloca API Management delante del webhook del agente alojado y usa identidades administradas, Key Vault y Cosmos DB para un historial de conversaciones duradero.

Note

La compatibilidad de Go con los agentes hospedados en Foundry estará disponible próximamente. Consulte el repositorio de Agent Framework Go para obtener el estado más reciente.

Tip

Consulte los ejemplos de Python o los ejemplos de C# para ver ejemplos de un proyecto de agente hospedado. O bien, use el comando azd ai agent init para generar la estructura de un nuevo proyecto de agente hospedado desde cero. Consulte esta guía de inicio rápido para obtener instrucciones paso a paso.

Ejecución local

La CLI de Azure Developer (azd) proporciona la manera más fácil de ejecutar y probar el agente hospedado localmente.

Inicialización de un proyecto

Cree una nueva carpeta e inicialice a partir de un manifiesto de ejemplo:

mkdir my-hosted-agent && cd my-hosted-agent
azd ai agent init -m <path-to-agent.manifest.yaml>

Tip

El manifiesto puede ser una ruta de acceso a un archivo YAML local o una dirección URL a un manifiesto remoto.

Establecimiento de variables de entorno

export FOUNDRY_PROJECT_ENDPOINT="https://<account>.services.ai.azure.com/api/projects/<project>"
export FOUNDRY_MODEL="<your-model-deployment>"

Ejecuta el host del agente

azd ai agent run

El host del agente se inicia en http://localhost:8088.

Invoque al agente

azd ai agent invoke --local "Hello!"

O bien, use curl:

curl -X POST http://localhost:8088/responses \
  -H "Content-Type: application/json" \
  -d '{"input": "Hello!"}'

O en PowerShell:

(Invoke-WebRequest -Uri http://localhost:8088/responses -Method POST -ContentType "application/json" -Body '{"input": "Hello!"}').Content

Implementación en Foundry

Una vez que haya comprobado el agente localmente, impleméntelo en Microsoft Foundry:

  1. Aprovisione recursos (si aún no tiene un proyecto Foundry):

    azd provision

    Esto crea un grupo de recursos con una instancia de Foundry, un proyecto, una implementación de modelos, Application Insights y un registro de contenedor.

  2. Implemente el agente:

    azd deploy

    Esto empaqueta al agente como una imagen de contenedor, lo inserta en Azure Container Registry e lo implementa en el servicio Foundry Agent.

La infraestructura de hospedaje de Foundry inserta automáticamente las siguientes variables de entorno en el contenedor del agente en tiempo de ejecución:

Variable Description
FOUNDRY_PROJECT_ENDPOINT Dirección URL del punto de conexión del proyecto Foundry.
AZURE_AI_MODEL_DEPLOYMENT_NAME Nombre de la implementación del modelo administrada por azd configurado en azd ai agent init. El código Python puede preferir FOUNDRY_MODEL localmente y recurrir a este valor hospedado.
APPLICATIONINSIGHTS_CONNECTION_STRING La cadena de conexión de Application Insights para la telemetría.

Una vez desplegado, el agente es accesible mediante su endpoint dedicado de Foundry y también se puede probar desde el portal de Foundry.

Pasos siguientes

Conceptos de agentes hospedados

  • Se actualizó por última vez el