Agents hébergés Foundry

Les agents hébergés dans Microsoft Foundry Agent Service vous permettent de déployer des applications d’agent conteneurisées sur une infrastructure gérée par Microsoft. La plateforme gère la mise à l’échelle, la persistance de l’état de session, la sécurité et la gestion du cycle de vie pour vous concentrer sur la logique de votre agent. Microsoft Les agents hébergés Foundry sont généralement disponibles et prennent en charge les agents créés avec votre propre code ou une infrastructure d’agent préférée. Cet article traite spécifiquement de l’intégration de l’hébergement agent Framework.

À l’aide de l’intégration de l’hébergement Agent Framework, vous pouvez exposer un Agent via le protocole Foundry Responses ou Invocations avec un code minimal. Python prend également en charge l’hébergement d’un composant natif directement Workflow, sans le convertir en agent.

Note

Vous pouvez également déployer le code de l’agent créé avec d’autres frameworks sur des agents hébergés par Foundry à l’aide de flux de travail CLI pour développeurs Azureazd. Pour obtenir des conseils sur les concepts et le déploiement indépendants de l’infrastructure, consultez Qu’est-ce que les agents hébergés ? Le reste de cet article se concentre sur l’intégration de Agent Framework.

Quand utiliser des agents hébergés

Choisissez les agents hébergés Foundry lorsque vous souhaitez :

  • Infrastructure managée : il n’est pas nécessaire de configurer des conteneurs, des serveurs web ou des règles de mise à l’échelle vous-même.
  • Gestion de session intégrée : la plateforme conserve et charge les $HOME fichiers entre les tours et les périodes d’inactivité.
  • Identité de l’agent dédié : chaque agent déployé obtient sa propre identité Entra pour un accès sécurisé aux modèles, outils et services en aval.
  • Points de terminaison compatibles OpenAI : les clients peuvent interagir avec votre agent à l’aide de n’importe quel KIT de développement logiciel (SDK) compatible OpenAI via le protocole Réponses.
  • Pour les agents audio en temps réel, utilisez des agents hébergés avec Azure Speech dans Foundry Tools (Voice Live) pour la détection d’activité vocale côté serveur, l’annulation d’écho et la réduction du bruit. Pour plus d’informations, consultez Utiliser Voice Live avec des agents hébergés.

Note

L’intégration Python agent-framework-foundry-hosting est en préversion. Microsoft Foundry Hosted Agents, le service d’hébergement géré, est désormais disponible de manière générale.

Prerequisites

Pour les tests locaux, vous avez également besoin des éléments suivants :

  • Un projet Microsoft Foundry avec un déploiement de modèle (par exemple, gpt-4o)
  • Azure CLI installé et authentifié (az login)

Installez le package NuGet d’hébergement :

dotnet add package Microsoft.Agents.AI.Foundry.Hosting --prerelease
  • Python 3.10 ou version ultérieure

Installez le package d’hébergement préversion, le client Foundry et Azure package d’authentification :

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

Dans Foundry, la plateforme fournit le contexte utilisateur et le contexte d’appel de l’appelant ; l’infrastructure d’hébergement les utilise pour isoler l’état par utilisateur et transférer le contexte de demande aux services Foundry. Les exécutions locales ne reçoivent pas ce contexte de plateforme. Les applications doivent donc fournir leurs propres contrôles d’identité et d’état si nécessaire.

Protocole des réponses

Le protocole Réponses est le point de départ recommandé pour la plupart des agents. Il expose un point de terminaison compatible /responses OpenAI et la plateforme gère automatiquement l’historique des conversations, la diffusion en continu et le cycle de vie des sessions.

Pour les agents Python hébergés, une réponse qui se termine prématurément a l’état incomplete. Les clients de streaming reçoivent un événement terminal response.incomplete, tandis que les clients sans streaming reçoivent status défini sur incomplete. Un motif de fin content_filter correspond à content_filter défini sur incomplete_details.reason, et length correspond à max_output_tokens. Tout contenu de sortie ou de refus généré reste disponible dans la réponse.

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("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 crée un hôte d’application préconfiguré pour l’environnement d’hébergement Foundry. AddFoundryResponses inscrit votre agent auprès du gestionnaire de protocole Réponses et MapFoundryResponses mappe le /responses point de terminaison 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["AZURE_AI_MODEL_DEPLOYMENT_NAME"],
    credential=DefaultAzureCredential(),
)

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

server = ResponsesHostServer(agent)
server.run()

ResponsesHostServer encapsule votre agent et l’expose via le protocole Foundry Responses. Le champ store de l’appelant contrôle si la réponse externe, la session gérée par l’hôte et l’état d’approbation sont enregistrés. Le history_source paramètre sélectionne indépendamment qui fournit l’historique des modèles :

history_source Comportement de l’historique des modèles
"agent_server" (valeur par défaut) L’hôte reconstruit la transcription des réponses externes stockées et désactive le stockage de service en aval pour empêcher l’historique des doublons.
"service" L’hôte envoie uniquement l’entrée actuelle et enregistre en privé l’ID de continuation du service de modèle de stockage. Une conversation de fournisseur enregistrée ne peut pas créer de branche à partir d’une réponse antérieure.
"agent" L’hôte envoie uniquement l’entrée actuelle. Les valeurs par défaut du stockage de l’agent HistoryProvider ou du service en aval gèrent l’historique.

Ne combinez pas "agent_server" ou "service" avec un HistoryProvider avec charge activée. Le mode par défaut rejette également les options de continuation en aval fixes telles que conversation_id, previous_response_idet conversation. Utiliser history_source="agent" pour une implémentation personnalisée SupportsAgentRun .

Le response_store paramètre de constructeur sélectionne le serveur principal pour la persistance des réponses externes. L’ancien paramètre du constructeur store est un alias déprécié de response_store ; aucun des deux paramètres ne définit le champ store de l’appelant, propre à chaque requête. Une requête avec store=false est à usage unique : elle n’enregistre pas l’état géré par l’hôte, désactive le stockage en aval pris en charge et ne peut pas utiliser background=true.

L’hôte possède l’agent fourni et peut ajouter des fournisseurs de contexte spécifiques à l’hébergement. Ne réutilisez pas l’agent sur un autre hôte et ne l’invoquez pas directement après avoir construit l’hôte.

L’hôte Responses préserve les appels système natifs, les captures d’écran et les vérifications de sécurité. Votre application doit exécuter les actions demandées et reconnaître explicitement les contrôles de sécurité. Pour obtenir le flux complet, consultez Utilisation de l’ordinateur natif.

Choisir une instance ou une fabrique d’agents

Les deux ResponsesHostServer et InvocationsHostServer acceptent une instance d’agent ou un argument égal à zéro ou asynchrone pouvant être appelé par le biais du agent paramètre. L’hôte réutilise une instance tout au long de sa durée de vie. Une fonction appelable s’exécute une fois par requête, et l’agent renvoyé appartient à cette requête.

Utilisez un callable lorsque l’agent conserve un état mutable en dehors de AgentSession. En particulier, créez une WorkflowAgent à partir d’une fabrique qui génère un nouveau workflow, des exécuteurs et des agents encapsulés :

def create_workflow_agent():
    return build_workflow().as_agent(name="support-workflow")


server = ResponsesHostServer(agent=create_workflow_agent)

Conservez le nom du flux de travail et les ID d’exécuteur stables afin que les demandes de réponses ultérieures puissent localiser les points de contrôle enregistrés. ResponsesHostServer maintient l’état pris en charge via ses stockages de session, de point de reprise et d’approbation de fonctions ; il ne conserve pas de champs arbitraires sur un agent à portée de requête. Consultez les exemples de flux de travaildurables et résilients .

Utilisez également une fabrique lorsqu’une intégration transporte l’identité de la requête ou possède des ressources propres à la requête. Par exemple, créez des connexions MCP, des boîtes à outils, des fournisseurs de compétences, des clients de recherche, des fournisseurs de mémoire et leurs informations d’identification à l’intérieur de la fabrique lorsqu’ils utilisent l’appel de plateforme actuel ou le contexte utilisateur. La réutilisation d’une connexion MCP à l’échelle du processus peut conserver l’identité de la demande qui l’a ouverte.

L’hôte entre et quitte les agents créés à l’usine pour chaque requête. Agent gère les clients gérés par le contexte et les outils MCP, mais votre fabrique doit fermer tout autre fournisseur, transport ou informations d’identification qu’elle crée. Ne fermez pas les objets partagés fournis par l’application en dehors de la fabrique.

Héberger un flux de travail natif avec Responses

Python peut héberger un flux de travail généré directement via workflow=. Un flux de travail natif nécessite une fonction de rappel parse_response qui associe la requête Responses actuelle à une entrée de démarrage typée ou au lot complet de réponses en attente :

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__}"],
    ),
)

Utilisez une fabrique synchrone ou asynchrone sensible au contexte de requête pour les workflows qui peuvent suspendre, continuer ou reprendre ou restaurer le travail en arrière-plan. La fabrique doit retourner un graphe nouvellement construit avec des exécuteurs, agents, clients, fournisseurs et outils nouveaux et mutables. Conservez le nom du flux de travail et les ID d’exécuteur stables afin que l’hôte puisse restaurer le point de contrôle exact associé à la réponse externe.

L’utilisateur de plateforme de confiance et le bac à sable Foundry isolent l’état du flux de travail natif. L’hôte valide un lot complet de réponses en attente avant qu’il n’utilise une quelconque autorisation de réponse. Les réponses obsolètes, partielles, dupliquées, rejouées, inter-utilisateurs et inter-bacs à sable échouent avant l’exécution du flux de travail. Une requête avec store=false n’enregistre pas l’état du flux de travail et ne peut pas retourner une pause pouvant être reprise.

Pour un flux de travail hérité qui accepte list[Message], utilisez response_input_messages(request) pour convertir uniquement le tour des réponses actuelles. Il ne charge pas l’historique extérieur précédent ou ne décode pas les réponses de flux de travail en attente. Hosting agent=workflow.as_agent() reste disponible pendant la version bêta actuelle, mais émet un avertissement de dépréciation. Pour obtenir des exemples complets, consultez les exemples de flux de travail native Responses.

Conserver l’état et gérer les conversations de longue durée

ResponsesHostServer et InvocationsHostServer configurez les magasins de sessions persistantes par défaut. AgentSessionStoreProvider fournit un FoundryAgentSessionStore ; les sessions de réponses utilisent le stockage logique agent_sessions, tandis que les sessions d’invocation utilisent le stockage distinct invocation_sessions. Ces espaces de stockage utilisent Foundry State Store en environnement hébergé, et le stockage sur fichier du SDK lors d’une exécution en local.

Pour les agents du workflow Responses, CheckpointStoreProvider fournit un FoundryCheckpointStore. Les flux de travail de réponses et d’appels natifs utilisent le même fournisseur pour leurs points de reprise exacts. FunctionApprovalStoreProvider fournit un FoundryFunctionApprovalStore pour les approbations d’outils d’agent en attente. Les réponses natives aux demandes et aux approbations de workflow sont plutôt liées aux points de contrôle du workflow.

Lors de l’exécution dans Foundry, l’interpréteur Python par défaut enregistre l’état de l’espace de noms en fonction de l’ID utilisateur de la plateforme et de l’ID de session de sandbox Foundry. Ils nécessitent également un ID d’appel de plateforme pour chaque opération d’état. L’ID d’appel autorise et met en corrélation l’opération ; Il ne s’agit pas d’un ID de conversation et ne fait pas partie de la clé de stockage.

Pour les réponses, le FOUNDRY_AGENT_SESSION_ID configuré par la plateforme identifie l’environnement sandbox, et un autre agent_session_id fourni par l’appelant est rejeté. Pour les invocations, l’hôte vérifie le paramètre de requête agent_session_id acheminé en le comparant au contexte de la requête. Si FOUNDRY_AGENT_SESSION_ID n’est pas configuré, le paramètre de requête doit être présent, non vide et correspondre au contexte de la requête. Les valeurs manquantes, dupliquées ou contradictoires sont rejetées au lieu d’utiliser un ID de secours du SDK.

Ces garanties s’appliquent aux magasins hébergés par défaut. Les fournisseurs de stockage personnalisés doivent mettre en œuvre une isolation équivalente des utilisateurs et du bac à sable, préserver le contenu interne AgentSession.session_id séparément des clés de recherche de l’hôte, et utiliser des écritures conditionnelles afin que des requêtes obsolètes ne puissent pas écraser des instantanés plus récents. Les nouvelles clés doivent utiliser des écritures de création uniquement plutôt que des opérations d’upsert inconditionnelles. Consultez l’exemple de stockage personnalisé pour une implémentation Cosmos DB avec des écritures et des suppressions protégées par ETag.

Avec history_source="agent", le magasin de sessions configuré conserve l’état du fournisseur transmis par AgentSession, y compris les messages de InMemoryHistoryProvider.

Les deux hôtes acceptent un StoreProvider[SessionStore] via agent_session_store_provider. L’état de session doit prendre en charge la sérialisation AgentSession. Enregistrer des codecs pour les types d’état personnalisés avec register_state_type() ; l’état restauré ne préserve pas l’identité des objets Python. Les nouveaux stockages par défaut font expirer les sessions 30 jours après leur dernière opération d’écriture. Les fournisseurs personnalisés gèrent leur propre politique de rétention des données.

Les magasins par défaut délimités ne lisent pas les données héritées non délimitées agent_sessions, invocation_sessions, de point de contrôle ou d’approbation de fonction. Commencez une nouvelle conversation Responses au lieu de réutiliser une ancienne previous_response_id ou un ancien identifiant de conversation. Les appels commencent avec une session Agent Framework vide dans le magasin délimité.

Les enregistrements AgentSession chargés utilisent des conditions ETag. Si une autre requête fait progresser la même session en premier, l’écriture obsolète échoue au lieu d’écraser un état plus récent. Cette vérification ne fournit ni transactions ni exécution unique pour les effets secondaires de l’agent ou de l’outil ; les applications doivent donc toujours coordonner les requêtes concurrentes.

Pour le stockage spécifique aux réponses, passez une StoreProvider à function_approval_store_provider ou une ContextScopedStoreProvider à checkpoint_store_provider.

Le traitement en arrière-plan externe utilise response.id, visible par l’appelant, pour effectuer l’interrogation. La valeur par défaut background_source="agent_server" conserve l’exécution en arrière-plan dans l’hôte. Définissez background_source="provider" uniquement avec history_source="service" et un client Responses avec stockage et reprise. Si ResponsesServerOptions(resilient_background=True) est également défini, l’hôte peut reprendre l’interrogation du fournisseur uniquement après avoir enregistré le jeton de continuation privé. Faites des effets secondaires de l’outil local idempotent, car un blocage avant l’enregistrement du jeton suivant peut les relire.

Importez ResponsesServerOptions depuis azure.ai.agentserver.responses, puis transmettez-le à options via le paramètre ResponsesHostServer. Les options de conversation longues disponibles dépendent du type d’agent :

Capacité Type d’agent Exigences et comportement
Récupération en arrière-plan du point de contrôle de flux de travail Flux de travail uniquement Définissez ResponsesServerOptions(resilient_background=True). Envoyez la demande Réponses avec store=true et background=true. Après un redémarrage, l’hôte reprend le dernier point de contrôle de flux de travail durable ou relecture l’entrée d’origine si aucun point de contrôle n’existe. Ne configurez pas le stockage de points de contrôle sur le flux de travail, car l’hôte le gère. Veillez à ce que les effets secondaires externes soient idempotents, car le traitement effectué après le dernier point de contrôle durable peut être répété.
Réponses en arrière-plan natives du fournisseur Sans workflow Agent avec un client stockant les réponses Définir history_source="service" et background_source="provider". Définir resilient_background=True lorsque les jetons de continuation du fournisseur enregistrés doivent être conservés après un redémarrage de l’hôte.
Conversations orientables Temporairement indisponible Ne définissez pas steerable_conversations=True. L’hôte génère RuntimeError lors de la construction tant que le SDK Agent Server ne gère pas en toute sécurité les instructions de pilotage rejetées.

Pour obtenir des implémentations complètes, consultez le stockage personnalisé, l’historique des réponses de base et l’arrière-plan et les exemples de flux de travail de longue durée résilients .

Lire des fichiers à partir du bac à sable hébergé

Traitez la persistance $HOME d’un bac à sable hébergé comme une ressource routée par requête, et non comme une limite générale du système de fichiers. N’acceptez que les fichiers que votre application téléverse explicitement dans un répertoire dédié, validez l’identité actuelle du bac à sable et rejetez les chemins absolus, les séquences de traversée de répertoires, les liens symboliques, les fichiers non ordinaires et le contenu surdimensionné ou non valide.

Pour le protocole Responses, acheminez une requête vers une session hébergée à l’aide du champ agent_session_id body. Le sélecteur de chaîne de requête s’applique à Invocations. Les chargements de session et les fichiers d’interpréteur de code de boîte à outils sont des ressources distinctes ; Un fichier bac à sable chargé n’est pas monté automatiquement dans un conteneur de boîte à outils. Consultez l’exemple de fichiers de session pour obtenir des lectures UTF-8 limitées et des instructions de chargement locales et hébergées.

Options de demande de contrôle

L’hôte mappe les champs de génération de réponses natives aux options d’exécution d’Agent Framework. Par exemple, max_output_tokens devient max_tokens et parallel_tool_calls devient allow_multiple_tool_calls. Les valeurs aplaties de extra_body remplacent les valeurs natives traduites.

Utilisez le hook synchrone ou asynchrone prepare_options(request, options) pour supprimer ou remplacer les options de modèle d’appelant avant l’exécution d’un agent standard. Le hook ne peut pas définir les champs d’identité, de stockage, de continuation ou de transport contrôlés par l’hôte. Pour une implémentation personnalisée SupportsAgentRun qui ne peut pas accepter les options de modèle d’exécution, définie unsupported_options sur "warn" (valeur par défaut), "ignore"ou "error".

Lorsqu’un outil MCP hébergé par Foundry nécessite le consentement de l’utilisateur, ResponsesHostServer retourne une réponse incomplète avec un oauth_consent_request élément de sortie. Présentez-le consent_link à l’utilisateur, puis poursuivez avec l’ID de la réponse incomplète, comme previous_response_id après le consentement de l’utilisateur. L’hôte conserve la session de l’agent pour cette nouvelle tentative et expose uniquement les liens de consentement HTTPS absolus.

Si votre hôte connaît les origines d’autorisation attendues, limitez les liens de consentement avec allowed_oauth_consent_origins:

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

Si aucune liste d’autorisation n’est spécifiée, la validation des URL HTTPS absolues est conservée sans restreindre l’origine de destination. La fourniture d’une liste vide rejette chaque lien de consentement. Configurer uniquement les origines HTTPS exactes ; les entrées avec un chemin d’accès, une requête ou un fragment sont rejetées.

Protocole d'invocation

Le protocole Invocations vous donne un contrôle total sur la requête et la réponse HTTP. Utilisez-le quand vous avez besoin de charges utiles personnalisées, de traitement non conversationnel ou de protocoles de streaming qui ne sont pas compatibles Avec OpenAI.

Avec le protocole Invocations en C#, vous implémentez un personnalisé InvocationHandler pour traiter les requêtes 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();

La AddInvocationsServer méthode inscrit les services de protocole Invocations. Vous implémentez InvocationHandler pour définir la façon dont votre agent traite chaque requête.

Pour une configuration légère, utilisez InvocationsHostServer du paquet agent_framework_foundry_hosting. Il encapsule votre agent de la même façon que ResponsesHostServer pour gérer automatiquement la gestion des sessions :

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["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 accepte les mêmes formes de fabrique d’instance ou à portée de requête que celles décrites pour l’hôte Responses. Il restaure les sessions sérialisées à partir du magasin configuré, afin que les conversations terminées puissent continuer après le redémarrage de l’hôte. Pour connaître le comportement de stockage, la rétention et la personnalisation, consultez l’état de persistance et gérer les conversations de longue durée.

Lorsqu’elles sont hébergées, les invocations utilisent la portée de requête vérifiée décrite dans Conserver l’état et gérer les conversations de longue durée. Traitez AgentSession.session_id comme une valeur opaque ; ne l’analysez pas et ne dépendez pas de sa représentation interne. Les exécutions locales conservent leur comportement de stockage mono-utilisateur existant.

Héberger un flux de travail natif avec Invocations

Transmettez workflow= et un rappel explicite parse_request pour héberger un flux de travail natif. Le callback définit le schéma JSON de l’application et retourne WorkflowTurn avec soit une entrée typée, soit le lot complet de réponses en attente :

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__}",
        ],
    ),
)

Incluez chaque type d’application personnalisé enregistré par le flux de travail dans la liste du fournisseur de points de allowed_checkpoint_types contrôle.

Les flux de travail hébergés nécessitent une fabrique sensible à la requête qui retourne un graphe nouvellement construit avec des ID de flux de travail et d’exécuteurs stables. Un workflow intégré directement est disponible uniquement pour une exécution locale et unique qui ne s’interrompt pas.

Les réponses de flux de travail sans diffusion en continu utilisent application/json avec une liste d’événements output . Le streaming émet des événements encadrés output et request_info, suivis de done uniquement après l’enregistrement du curseur du workflow exact. Traitez la sortie diffusée en continu comme provisoire jusqu’à done. Les flux de travail natifs ne prennent pas en charge legacy_wire_format=True.

L’hôte valide les réponses par rapport au point de contrôle exact en attente dans la portée de l’utilisateur de confiance et du bac à sable. Si un flux de travail a plusieurs demandes en attente, répondez au lot complet en une seule fois. Pour obtenir un analyseur exécutable, un flux de travail de ticket typé, une liste d’autorisations de types de points de contrôle et des exemples JSON/SSE, consultez l’exemple de workflow native Invocations.

Personnaliser les requêtes et les réponses des invocations

Par défaut, POST /invocations accepte un objet JSON avec une chaîne message, un objet facultatif options et une valeur booléenne stream facultative. Pour accepter une charge utile spécifique à l’application, transmettez un rappel synchrone ou asynchrone parse_request qui retourne InvocationRun(messages, options, stream). Utilisez prepare_options pour filtrer ou remplacer une copie des options de génération de l’appelant avant que l’agent ne s’exécute.

L’hôte valide le résultat du hook et rejette les contrôles relatifs à l’identité de la plateforme, au stockage, à la continuation et à l’exécution de l’agent. Pour les agents qui n’acceptent pas les options d’exécution, définis unsupported_options sur "warn" (valeur par défaut), "ignore"ou "error". Consultez l’exemple d’analyseur Invocations pour une implémentation complète.

En cas de succès sans streaming, la réponse est renvoyée au format JSON sous la forme {"response": "..."}. La diffusion en continu utilise des événements envoyés par le serveur : une ou plusieurs event: delta images, suivies d’une event: done réussite ou event: error d’un échec. Un flux peut émettre des deltas avant qu’une erreur ne survienne ; les clients doivent donc considérer done, et non un delta, comme une fin réussie. L’hôte émet done uniquement après qu’il finalise le flux de réponse et conserve le AgentSession. Son session_id correspond à l’ID d’itinéraire du bac à sable de la plateforme, et non au AgentSession.session_id sérialisé.

Définissez legacy_wire_format=True uniquement lors de la migration de clients existants qui nécessitent la réponse en texte brut précédent et le flux de blocs de texte brut. Ce mode de compatibilité est déconseillé et ne convertit pas les échecs en texte réussi. L’hôte sérialise les demandes de même session au sein d’un seul processus ; Un conflit de comparaison et d’échange interprocesseurs peut toujours se produire après les effets de l’outil externe.

Le protocole Invocations ne reprend pas les exécutions de flux de travail en attente ou interrompues. Utilisez le modèle de gestionnaire personnalisé dans la section suivante lorsque vous avez besoin d’un comportement de continuation de flux de travail différent.

Pour un contrôle total sur la gestion des demandes, utilisez directement InvocationAgentServerHost du paquet azure.ai.agentserver.invocations et implémentez votre propre gestionnaire d’appels :

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["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()

Avertissement

Le magasin de sessions en mémoire dans l’exemple de gestionnaire personnalisé est perdu lors du redémarrage. Utilisez un stockage durable (par exemple, Cosmos DB) en production.

Pour un déploiement complet d’Invocations, consultez l’exemple Telegram hébergé sur Foundry. Il place API Management devant le webhook de l’agent hébergé et utilise des identités managées, Key Vault et Cosmos DB pour assurer un historique persistant des conversations.

Note

La prise en charge de Go pour les agents hébergés de Foundry sera bientôt disponible. Consultez le référentiel Agent Framework Go pour connaître l’état le plus récent.

Tip

Reportez-vous aux exemples Python ou aux exemples C# pour obtenir des exemples de projet d’agent hébergé. Vous pouvez également utiliser la azd ai agent init commande pour générer automatiquement un nouveau projet d’agent hébergé à partir de zéro. Reportez-vous à ce guide de démarrage rapide pour obtenir des instructions pas à pas.

Exécution locale

L’interface CLI Azure développeur (azd) offre le moyen le plus simple d’exécuter et de tester votre agent hébergé localement.

Initialiser un projet

Créez un dossier et initialisez à partir d’un exemple de manifeste :

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

Tip

Le manifeste peut être un chemin d’accès à un fichier YAML local ou à une URL vers un manifeste distant.

Définir des variables d’environnement

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

Exécuter l’hôte de l’agent

azd ai agent run

L’hôte de l’agent démarre sur http://localhost:8088.

Appeler l’agent

azd ai agent invoke --local "Hello!"

Ou utilisez curl:

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

Ou dans PowerShell :

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

Déploiement sur Foundry

Une fois que vous avez vérifié votre agent localement, déployez-le sur Microsoft Foundry :

  1. Provisionnez des ressources (si vous n’avez pas encore de projet Foundry) :

    azd provision

    Cela crée un groupe de ressources avec une instance Foundry, un projet, un déploiement de modèle, Application Insights et un registre de conteneurs.

  2. Déployez l’agent :

    azd deploy

    Cela empaquette votre agent en tant qu'image conteneur, l'envoie à Azure Container Registry et le déploie dans Foundry Agent Service.

L’infrastructure d’hébergement Foundry injecte automatiquement les variables d’environnement suivantes dans votre conteneur d’agent au moment de l’exécution :

Variable Description
FOUNDRY_PROJECT_ENDPOINT URL du point de terminaison du projet Foundry.
AZURE_AI_MODEL_DEPLOYMENT_NAME Nom du déploiement du modèle (configuré pendant azd ai agent init).
APPLICATIONINSIGHTS_CONNECTION_STRING La chaîne de connexion Application Insights pour la télémétrie.

Une fois déployé, votre agent est accessible via son point de terminaison Foundry dédié et peut également être testé à partir du portail Foundry.

Étapes suivantes

Concepts relatifs aux agents hébergés

  • Last updated on