Agenti ospitati di Foundry

• .NET 10 SDK o versione successiva

Installare il pacchetto NuGet di hosting:

dotnet add package Microsoft.Agents.AI.Foundry.Hosting --prerelease

• Python 3.10 o versione successiva

Installare il pacchetto di hosting prerelease, il client Foundry e il pacchetto di autenticazione di Azure:

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

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 dell'applicazione preconfigurato per l'ambiente host Foundry. AddFoundryResponses registra l'agente con il gestore del protocollo Responses ed MapFoundryResponses esegue il mapping dell'endpoint /responses 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()

L'elemento ResponsesHostServer esegue il wrapping dell'agente e lo espone tramite il protocollo Foundry Responses. Il campo store del chiamante controlla se vengono salvati la risposta esterna, la sessione gestita dall'host e lo stato di approvazione. L'impostazione history_source seleziona in modo indipendente chi fornisce la cronologia dei modelli:

Non combinare "agent_server" o "service" con un HistoryProvider abilitato al caricamento. La modalità predefinita rifiuta anche le opzioni di continuazione predefinite a valle come conversation_id, previous_response_id e conversation. Usare history_source="agent" per un'implementazione personalizzata SupportsAgentRun .

Il response_store parametro del costruttore seleziona il back-end per la persistenza delle risposte esterne. Il parametro store del costruttore precedente è un alias deprecato per response_store; nessuno dei due parametri imposta il campo per richiesta store del chiamante. Una richiesta con store=false è un'unica operazione: non salva lo stato gestito dall'host, disabilita l'archiviazione downstream supportata e non può usare background=true.

L'host è proprietario dell'agente fornito e potrebbe aggiungere provider di contesto specifici dell'hosting. Non riutilizzare l'agente con un altro host né invocarlo direttamente dopo la creazione dell'host.

L'host Responses preserva le chiamate di sistema, gli screenshot e i controlli di sicurezza del sistema locale. L'applicazione deve eseguire le azioni richieste e confermare in modo esplicito eventuali controlli di sicurezza. Per il flusso completo, vedere Uso del computer nativo.

Scegliere un'istanza o una factory dell'agente

Sia ResponsesHostServer che InvocationsHostServer accettano un'istanza dell'agente o un argomento zero sincrono o asincroni chiamabili tramite il agent parametro . L'host riutilizza un'istanza per tutto il suo ciclo di vita. Una funzione callable viene eseguita una sola volta per richiesta e l'agente restituito appartiene a quella richiesta.

Usare un callable quando un agente regolare mantiene uno stato modificabile al di fuori di AgentSession. Gli host persistono solo i relativi archivi di sessione, checkpoint e approvazione delle funzioni supportati, non campi arbitrari in un agente con ambito richiesta.

Usare anche una factory quando un'integrazione contiene l'identità della richiesta o possiede risorse specifiche della richiesta. Ad esempio, creare connessioni MCP, Toolbox, provider di skill, client Search, provider di memoria e le relative credenziali all'interno della factory quando usano la chiamata alla piattaforma o il contesto utente corrente. Il riutilizzo di una connessione MCP a livello di processo può mantenere l'identità della richiesta che l'ha aperta.

L'host accede agli agenti creati dalla factory e ne esce per ogni richiesta. Agent gestisce i client gestiti dal contesto e gli strumenti MCP, ma la factory deve chiudere qualsiasi altro provider, trasporto o credenziale creato. Non chiudere gli oggetti condivisi che l'applicazione ha fornito dall'esterno della factory.

Gestisci un flusso di lavoro nativo con Responses

Python può ospitare un workflow creato direttamente tramite workflow=. Un flusso di lavoro nativo richiede un parse_response callback che esegue il mapping della richiesta di risposte corrente a un input iniziale tipizzato o al batch completo di risposte in sospeso:

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

Usare una factory sincrona o asincrona compatibile con le richieste per i flussi di lavoro che possono sospendere, continuare o ripristinare il lavoro in background. La fabbrica deve restituire un grafo appena creato con nuovi executor modificabili, agenti, client, provider e strumenti. Mantenere stabili il nome del flusso di lavoro e gli ID executor in modo che l'host possa ripristinare il checkpoint esatto associato alla risposta esterna.

L'utente trusted platform e la sandbox Foundry isolano lo stato del flusso di lavoro nativo. L'host convalida un batch di risposta in sospeso completo prima di utilizzare qualsiasi autorizzazione alla risposta. Le risposte non aggiornate, parziali, duplicate, riutilizzate tramite replay, tra utenti diversi e cross-sandbox hanno esito negativo prima dell'esecuzione del flusso di lavoro. Una richiesta con store=false non salva lo stato del flusso di lavoro e non può restituire una pausa riprendibile.

Per un flusso di lavoro legacy che accetta list[Message], usare response_input_messages(request) per convertire solo il turno Risposte corrente. Non carica la cronologia esterna precedente o non decodifica le risposte del flusso di lavoro in sospeso. L'hosting agent=workflow.as_agent() rimane disponibile durante la versione beta corrente, ma genera un avviso di deprecazione. Per esempi completi, vedere gli esempi nativi del flusso di lavoro Responses.

Mantenere lo stato e gestire conversazioni di lunga durata

ResponsesHostServer e InvocationsHostServer configurano per impostazione predefinita gli archivi permanenti delle sessioni. AgentSessionStoreProvider fornisce un oggetto FoundryAgentSessionStore; Le sessioni di risposte usano l'archivio agent_sessions logico, mentre le sessioni chiamate usano l'archivio separato invocation_sessions . Questi archivi usano Foundry State Store quando sono ospitati e l'archiviazione basata su file dell'SDK quando vengono eseguiti localmente.

Per gli agenti del flusso di lavoro Responses, CheckpointStoreProvider fornisce un FoundryCheckpointStore. I flussi di lavoro Risposte native e Invocazioni usano lo stesso provider per i checkpoint esatti per la continuazione. FunctionApprovalStoreProvider fornisce un elemento FoundryFunctionApprovalStore per le approvazioni dello strumento agente in sospeso. Le risposte di richiesta e approvazione del flusso di lavoro native sono invece associate ai checkpoint del flusso di lavoro.

Quando viene eseguito in Foundry, l'ambiente Python predefinito memorizza lo stato del namespace in funzione dell'ID utente della piattaforma e dell'ID della sessione sandbox di Foundry. Richiedono anche un ID di chiamata della piattaforma per ogni operazione di stato. L'ID chiamata autorizza e correla l'operazione; non è un ID conversazione e non fa parte della chiave di archiviazione.

Per le risposte, il valore FOUNDRY_AGENT_SESSION_ID configurato dalla piattaforma identifica la sandbox e un valore agent_session_id diverso fornito dal chiamante viene rifiutato. Per le invocazioni, l'host verifica il parametro di query agent_session_id instradato rispetto al contesto della richiesta. Se FOUNDRY_AGENT_SESSION_ID non è configurato, il parametro di query deve essere presente, nonempty e corrispondere al contesto della richiesta. I valori mancanti, duplicati o in conflitto vengono rifiutati invece di usare un ID di fallback SDK.

Queste garanzie si applicano agli store ospitati predefiniti. I provider di archiviazione personalizzati devono implementare un isolamento equivalente degli utenti e delle sandbox, preservare l'elemento interno AgentSession.session_id separato dalle chiavi di ricerca dell'host e utilizzare scritture condizionali affinché le richieste obsolete non possano sovrascrivere gli snapshot più recenti. Le nuove chiavi dovrebbero usare operazioni di sola creazione anziché upsert incondizionati. Vedere l'esempio di archiviazione personalizzata per un'implementazione di Cosmos DB con scritture ed eliminazioni protette da ETag.

Con history_source="agent", l'archivio sessioni configurato mantiene lo stato del provider trasportato da AgentSession, inclusi i messaggi da InMemoryHistoryProvider.

Entrambi gli host accettano da StoreProvider[SessionStore] a agent_session_store_provider. Lo stato della sessione deve supportare AgentSession la serializzazione. Registrare i codec per i tipi di stato personalizzati con register_state_type(); lo stato ripristinato non preserva l'identità dell'oggetto Python. I nuovi archivi dati predefiniti fanno scadere le sessioni 30 giorni dopo l'ultima operazione di scrittura. I fornitori personalizzati gestiscono autonomamente i propri criteri di conservazione.

Gli archivi predefiniti con ambito non leggono i dati legacy senza ambito agent_sessions, invocation_sessions, di checkpoint o di approvazione delle funzioni. Avvia una nuova conversazione Responses invece di riutilizzare un vecchio previous_response_id o un ID conversazione. Le invocazioni iniziano con una sessione vuota di Agent Framework nell'archivio con ambito definito.

I record caricati AgentSession usano condizioni ETag. Se un'altra richiesta fa avanzare per prima la stessa sessione, la scrittura obsoleta fallisce invece di sovrascrivere lo stato più recente. Questo controllo non fornisce transazioni o esecuzione esattamente una sola volta per gli effetti collaterali dell'agente o degli strumenti, quindi le applicazioni devono comunque coordinare le richieste sovrapposte.

Per l'archiviazione specifica di Responses, passa un oggetto StoreProvider a function_approval_store_provider oppure un oggetto ContextScopedStoreProvider a checkpoint_store_provider.

Il lavoro in background esterno usa response.id, visibile al chiamante, per il polling. Il valore predefinito background_source="agent_server" mantiene l'esecuzione in background nell'host. Impostare background_source="provider" solo con history_source="service" e un client di risposte di archiviazione e ripristinabile. Se anche ResponsesServerOptions(resilient_background=True) è impostato, l'host può riprendere il polling del provider solo dopo aver salvato il token di continuazione privato. Rendere idempotenti gli effetti collaterali dello strumento locale perché un arresto anomalo prima che il token successivo venga salvato può riprodurre tali effetti.

Importare ResponsesServerOptions da azure.ai.agentserver.responsese passarlo al ResponsesHostServeroptions parametro . Le opzioni di conversazione a esecuzione prolungata disponibili dipendono dal tipo di agente:

Per le implementazioni complete, vedere l'archivio personalizzato, la cronologia delle risposte di base e gli esempi di flusso di lavoro resilienti a esecuzione prolungata .

Leggere i file dalla sandbox ospitata

Tratta l'elemento persistente $HOME di una sandbox ospitata come una risorsa instradata tramite richiesta, non come un confine generale del file system. Accettare solo i file caricati in modo esplicito dall'applicazione in una directory dedicata, convalidare l'identità sandbox corrente e rifiutare percorsi assoluti, attraversamento, collegamenti, file nonregulari e contenuto sovradimensionato o non valido.

Per il protocollo Responses, instradare una richiesta a una sessione ospitata tramite il campo agent_session_id body. Il selettore della stringa di query è per le invocazioni. I caricamenti di sessione e i file dell'interprete del codice della casella degli strumenti sono risorse separate; un file sandbox caricato non viene montato automaticamente in un contenitore della casella degli strumenti. Vedi l’esempio dei file di sessione per le letture UTF-8 con limite e le indicazioni per il caricamento locale e su host.

Opzioni di richiesta di controllo

L'host mappa i campi di generazione nativi di Responses alle opzioni di esecuzione di Agent Framework. Ad esempio, max_output_tokens diventa max_tokens, e parallel_tool_calls diventa allow_multiple_tool_calls. I valori appiattiti di extra_body sovrascrivono i valori nativi tradotti.

Usa l'hook sincrono o asincrono prepare_options(request, options) per rimuovere o sostituire le opzioni del modello del chiamante prima che venga eseguito un agente normale. L'hook non può impostare i campi relativi a identità, archiviazione, continuazione o trasporto controllati dall'host. Per un'implementazione personalizzata SupportsAgentRun che non può accettare opzioni del modello in fase di esecuzione, impostare unsupported_options su "warn" (predefinito), "ignore" o "error".

Gestire le richieste di consenso OAuth

Quando uno strumento MCP ospitato in Foundry richiede il consenso dell'utente, ResponsesHostServer restituisce una risposta incompleta con un oauth_consent_request elemento di output. Presentare consent_link all'utente, quindi continuare usando l'ID della risposta incompleta come previous_response_id dopo che l'utente ha fornito il consenso. L'host mantiene la sessione dell'agente per questo nuovo tentativo ed espone solo collegamenti di consenso HTTPS assoluti.

Se l'host conosce le origini di autorizzazione previste, limitare i collegamenti di consenso con allowed_oauth_consent_origins:

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

L'assenza dell'allowlist mantiene la validazione HTTPS assoluta senza limitare l'origine di destinazione. Se si specifica un elenco vuoto, ogni collegamento di consenso viene rifiutato. Configurare solo le origini HTTPS esatte; le voci con un percorso, una query o un frammento vengono rifiutate.

Con il protocollo Invocations in C#, si implementa un InvocationHandler personalizzato per elaborare le richieste in ingresso.

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();

Il AddInvocationsServer metodo registra i servizi del protocollo Invocations. Si implementa InvocationHandler per definire il modo in cui l'agente elabora ogni richiesta.

Per un'installazione leggera, usare InvocationsHostServer dal agent_framework_foundry_hosting pacchetto. Esegue il wrapping dell'agente in modo analogo a ResponsesHostServer e gestisce automaticamente la gestione delle sessioni:

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 accetta la stessa istanza o le stesse forme factory con ambito di richiesta descritte per l'host Responses. Ripristina le sessioni serializzate dall'archivio configurato, in modo che le conversazioni completate possano continuare dopo il riavvio dell'host. Per il comportamento di memorizzazione, la conservazione e la personalizzazione, vedi Mantenere lo stato persistente e gestire conversazioni di lunga durata.

Quando è ospitato, Invocations usa l'ambito della richiesta verificato descritto in Rendere persistente lo stato e gestire conversazioni di lunga durata. Considera AgentSession.session_id come un valore opaco. Non analizzare o dipendere dalla relativa rappresentazione interna. Le esecuzioni locali mantengono il comportamento di archiviazione a utente singolo esistente.

Gestire un flusso di lavoro nativo con invocazioni

Passa workflow= e un callback esplicito parse_request per ospitare un flusso di lavoro nativo. Il callback è proprietario dello schema JSON dell'applicazione e restituisce un WorkflowTurn oggetto con input tipizzato o il batch di risposta in sospeso completo:

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

Includi ogni tipo di applicazione personalizzato che il flusso di lavoro salva nell'elenco del provider di checkpoint allowed_checkpoint_types.

I flussi di lavoro ospitati richiedono una factory sensibile al contesto della richiesta che restituisce un grafo appena creato con ID di flusso di lavoro ed executor stabili. Un flusso di lavoro creato direttamente è disponibile solo per un'esecuzione singola locale che non va in pausa.

Le risposte del flusso di lavoro non in streaming usano application JSON con un elenco di eventi output. Lo streaming genera eventi request_info e output incorniciati, seguiti da done solo dopo che il cursore esatto del flusso di lavoro è stato salvato. Considera l'output trasmesso come provvisorio fino a done. I flussi di lavoro nativi non supportano legacy_wire_format=True.

L'host convalida le risposte rispetto al checkpoint esatto in sospeso nell'ambito dell'utente attendibile e della sandbox. Se un flusso di lavoro ha più richieste in sospeso, rispondere al batch completo in un solo turno. Per un parser eseguibile, il flusso di lavoro del ticket tipizzato, l'elenco di elementi consentiti del tipo di checkpoint e gli esempi JSON/SSE, vedere l'esempio di flusso di lavoro di Invocations native.

Personalizzare le richieste e le risposte delle invocazioni

Per impostazione predefinita, POST /invocations accetta un oggetto JSON con una stringa message, un oggetto facoltativo e un valore booleano stream facoltativooptions. Per accettare un payload specifico dell'applicazione, passare un callback sincrono o asincrono parse_request che restituisce InvocationRun(messages, options, stream). Usare prepare_options per filtrare o sostituire una copia delle opzioni di generazione del chiamante prima dell'esecuzione dell'agente.

L'host convalida l'output dell'hook e rifiuta i controlli di identità della piattaforma, archiviazione, continuazione e esecuzione dell'agente. Per gli agenti che non accettano le opzioni di esecuzione, impostare "warn" su unsupported_options (predefinito), "ignore" o "error". Vedere l'esempio del parser di invocazioni per un'implementazione completa.

In caso di esito positivo senza streaming, viene restituito un JSON nella forma {"response": "..."}. Lo streaming utilizza eventi inviati dal server: uno o più frame event: delta, seguiti da event: error in caso di successo o da event: done in caso di errore. Un flusso può generare delta prima di un errore, pertanto i client devono considerare done, non un delta, come completamento riuscito. L'host emette done solo dopo aver finalizzato il flusso di risposta e aver reso persistente AgentSession. Il suo session_id è l'ID della route sandbox della piattaforma, non il AgentSession.session_id serializzato.

Imposta legacy_wire_format=True solo durante la migrazione di client esistenti che richiedono la precedente risposta in testo semplice e il flusso grezzo di blocchi di testo. Questa modalità di compatibilità è deprecata e non converte gli errori in testo riuscito. L'host serializza le richieste della stessa sessione solo all'interno di un processo; un conflitto tra processi di confronto e scambio può verificarsi ancora dopo gli effetti degli strumenti esterni.

Il protocollo Invocations non riprende le esecuzioni del workflow in sospeso o interrotte. Usare il modello di gestore personalizzato nella sezione seguente quando è necessario un comportamento di continuazione del flusso di lavoro diverso.

Per il controllo completo sulla gestione delle richieste, usare InvocationAgentServerHost direttamente dal azure.ai.agentserver.invocations pacchetto e implementare il proprio gestore invoke:

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

Per una distribuzione completa di Invocations, consulta l'esempio Telegram ospitato su Foundry. Colloca API Management davanti al webhook dell'agente ospitato e usa identità gestite, Key Vault e Cosmos DB per una cronologia delle conversazioni persistente.