In der Foundry gehostete Agents

• .NET 10 SDK oder höher

Installieren Sie das Hosting-NuGet-Paket:

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

• Python 3.10 oder höher

Installieren Sie das Vorabversionshostingpaket, den Foundry-Client und Azure Authentifizierungspaket:

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

Der AgentHost.CreateBuilder erstellt einen Anwendungshost, der für die Foundry-Hostingumgebung vorkonfiguriert ist. AddFoundryResponses registriert Ihren Agent mit dem Antwortprotokollhandler und MapFoundryResponses ordnet den /responses HTTP-Endpunkt zu.

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 umschließt Ihren Agenten und macht ihn über das Foundry Responses-Protokoll verfügbar. Das store Feld des Aufrufers steuert, ob die äußere Antwort sowie die hostverwaltete Sitzung und der Genehmigungsstatus gespeichert werden. Die history_source Einstellung legt unabhängig fest, wer den Modellverlauf bereitstellt:

Kombinieren Sie "agent_server" oder "service" nicht mit einem lastfähigen HistoryProvider. Der Standardmodus lehnt außerdem feste nachgelagerte Fortsetzungsoptionen wie conversation_id, previous_response_id und conversation ab. Verwenden Sie SupportsAgentRun für eine benutzerdefinierte history_source="agent"-Implementierung.

Der response_store Konstruktorparameter wählt das Back-End für die Persistenz der äußeren Antworten aus. Der ältere Konstruktorparameter store ist ein veralteter Alias für response_store; keiner der Parameter legt das Feld pro Anforderung store des Aufrufers fest. Eine Anfrage mit store=false ist eine einmalige Anfrage: Sie speichert keinen vom Host verwalteten Zustand, deaktiviert die unterstützte nachgelagerte Speicherung und kann background=true nicht verwenden.

Der Host besitzt den bereitgestellten Agent und fügt möglicherweise hostingspezifische Kontextanbieter hinzu. Verwenden Sie den Agent nicht für einen anderen Host wieder, und rufen Sie ihn nicht direkt nach dem Erstellen des Hosts auf.

Der Antworthost behält systemeigene Computeranrufe, Screenshots und Sicherheitsprüfungen bei. Ihre Anwendung muss die angeforderten Aktionen ausführen und alle Sicherheitsprüfungen explizit bestätigen. Den vollständigen Ablauf finden Sie unter "Native Computerverwendung".

Wählen Sie eine Agentinstanz oder eine Fabrik aus

Sowohl ResponsesHostServer als auch InvocationsHostServer akzeptieren über den Parameter agent entweder eine Agent-Instanz oder eine synchrone bzw. asynchrone aufrufbare Funktion ohne Argumente. Der Host verwendet eine Instanz während seiner gesamten Lebensdauer wieder. Eine aufrufbare Funktion wird pro Anfrage einmal ausgeführt, und der zurückgegebene Agent gehört zu dieser Anfrage.

Verwenden Sie ein Callable, wenn ein normaler Agent einen veränderbaren Zustand außerhalb AgentSession beibehält. Die Hosts speichern dauerhaft nur ihre unterstützten Speicher für Sitzungen, Checkpoints und Funktionsfreigaben, nicht jedoch beliebige Felder bei einem Agenten im Anforderungskontext.

Verwenden Sie auch eine Factory, wenn eine Integration eine Anfrageidentität mitführt oder anfragespezifische Ressourcen verwaltet. Legen Sie beispielsweise MCP-Verbindungen, Toolboxen, Skillanbieter, Such-Clients, Speicheranbieter und deren Anmeldeinformationen in der Factory an, wenn diese den aktuellen Plattformaufruf oder Benutzerkontext verwenden. Durch die erneute Nutzung einer prozessweiten MCP-Verbindung kann die Identität der Anforderung beibehalten werden, die sie geöffnet hat.

Der Host tritt für jede Anfrage in die durch die Factory erstellten Agenten ein und verlässt sie wieder. Agent verwaltet Clients mit Kontextverwaltung und MCP-Tools, Ihre Factory muss jedoch jeden anderen Provider, Transport oder Anmeldedaten schließen, den bzw. die sie erstellt. Schließen Sie keine gemeinsam genutzten Objekte, die die Anwendung außerhalb der Factory bereitgestellt hat.

Einen nativen Workflow mit Responses hosten

Python kann einen erstellten Workflow direkt über workflow= hosten. Ein nativer Workflow erfordert einen parse_response Rückruf, der die aktuelle Responses-Anfrage entweder einer typisierten Starteingabe oder dem vollständigen Batch ausstehender Antworten zuordnet:

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

Verwenden Sie eine anforderungsorientierte synchrone oder asynchrone Factory für Workflows, mit denen sich Hintergrundprozesse anhalten, fortsetzen oder wiederherstellen lassen. Die Factory muss einen neu erstellten Graphen mit neuen, veränderbaren Exekutoren, Agents, Clients, Anbietern und Tools zurückgeben. Lassen Sie den Workflownamen und die Executor-IDs stabil, damit der Host den genauen Checkpoint wiederherstellen kann, der der äußeren Antwort zugeordnet ist.

Der Trusted Platform-Benutzer und die Foundry-Sandbox isolieren den nativen Workflow-Zustand. Der Host überprüft einen vollständigen ausstehenden Antwortbatch, bevor er eine Antwortberechtigung in Anspruch nimmt. Veraltete, unvollständige, doppelte, wiederholte, benutzerübergreifende und sandboxübergreifende Antworten werden bereits vor der Ausführung des Workflows abgelehnt. Eine Anforderung mit store=false speichert den Workflowstatus nicht und kann keine fortsetzbare Pause zurückgeben.

Verwenden Sie für einen Legacyworkflow, der response_input_messages(request) akzeptiert, list[Message], um nur den aktuellen Responses-Turn zu konvertieren. Es wird keine vorherige externe Historie geladen oder ausstehende Workflowantworten decodiert. Das Hosting von agent=workflow.as_agent() bleibt während der aktuellen Betaversion verfügbar, gibt jedoch eine Veraltungswarnung aus. Vollständige Beispiele finden Sie in den Workflowbeispielen für native Responses.

Status beibehalten und zeitintensive Unterhaltungen verarbeiten

ResponsesHostServer und InvocationsHostServer konfigurieren standardmäßig persistente Sitzungsspeicher. AgentSessionStoreProvider stellt FoundryAgentSessionStore bereit; Antwortsitzungen verwenden den agent_sessions logischen Speicher, während Aufrufsitzungen den separaten invocation_sessions Speicher verwenden. Diese Speicher verwenden den Foundry State Store, wenn sie gehostet werden, und den dateigestützten Speicher des SDK, wenn Sie lokal ausgeführt werden.

Für Responses-Workflow-Agenten stellt CheckpointStoreProvider ein FoundryCheckpointStore bereit. Native Responses and Invocations-Workflows verwenden denselben Provider für ihre exakten Checkpoints für die Fortsetzung. FunctionApprovalStoreProvider stellt eine FoundryFunctionApprovalStore für ausstehende Agent-Toolgenehmigungen bereit. Native Workflowanforderungs- und Genehmigungsantworten sind stattdessen an Workflowprüfpunkte gebunden.

Bei der Ausführung in Foundry speichert die standardmäßige Python-Umgebung den Namespace-Status anhand der Plattformbenutzer-ID und der Foundry-Sandbox-Sitzungs-ID. Außerdem ist für jeden Statusvorgang eine Plattformaufruf-ID erforderlich. Die Anruf-ID autorisiert und korreliert den Vorgang; es handelt sich nicht um eine Unterhaltungs-ID und ist nicht Teil des Speicherschlüssels.

Für Antworten identifiziert die von der Plattform konfigurierte FOUNDRY_AGENT_SESSION_ID die Sandbox, und eine andere vom Aufrufer bereitgestellte agent_session_id wird abgelehnt. Bei Aufrufen überprüft der Host den Routingabfrageparameter agent_session_id anhand des Anforderungskontexts. Wenn FOUNDRY_AGENT_SESSION_ID nicht konfiguriert ist, muss der Abfrageparameter vorhanden sein, darf nicht leer sein und muss mit dem Anfragekontext übereinstimmen. Fehlende, doppelte oder widersprüchliche Werte werden abgelehnt, anstatt eine SDK-Fallback-ID zu verwenden.

Diese Garantien gelten für die standardmäßig gehosteten Stores. Benutzerdefinierte Store-Anbieter müssen eine gleichwertige Benutzer- und Sandbox-Isolation implementieren, das innere AgentSession.session_id getrennt von Host-Lookup-Schlüsseln beibehalten und bedingte Schreibvorgänge verwenden, damit veraltete Anfragen keine neueren Snapshots überschreiben können. Neue Schlüssel sollten Schreibvorgänge verwenden, die nur beim Erstellen erfolgen, anstatt unbedingte Upserts zu verwenden. Sehen Sie sich das benutzerdefinierte Speicherbeispiel für eine Cosmos DB-Implementierung mit ETag-geschützten Schreib- und Löschvorgängen an.

Mit history_source="agent" speichert der konfigurierte Sitzungsspeicher den von AgentSession mitgeführten Provider-Zustand, darunter Nachrichten von InMemoryHistoryProvider.

Beide Hosts akzeptieren StoreProvider[SessionStore] bis agent_session_store_provider. Der Sitzungszustand muss die AgentSession-Serialisierung unterstützen. Registrieren Sie Codecs für benutzerdefinierte Zustandstypen mit register_state_type(); der wiederhergestellte Zustand behält die Objektidentität von Python nicht bei. Neue Standard-Stores lassen Sitzungen 30 Tage nach dem letzten Schreibzugriff ablaufen. Benutzerdefinierte Anbieter legen ihre Aufbewahrungsfristen selbst fest.

Die standardmäßigen Speicher mit Geltungsbereich lesen keine älteren Daten ohne Geltungsbereich (agent_sessions, invocation_sessions), Checkpoint-Daten oder Daten zur Funktionsgenehmigung. Starten Sie eine neue Responses-Konversation, anstatt eine alte previous_response_id oder Konversations-ID wiederzuverwenden. Aufrufe beginnen mit einer leeren Agent Framework-Sitzung im bereichsbezogenen Speicher.

Geladene AgentSession Datensätze verwenden ETag-Bedingungen. Wenn eine andere Anforderung zuerst dieselbe Sitzung fortschreitet, schlägt der veraltete Schreibvorgang fehl, anstatt den neueren Zustand zu überschreiben. Diese Überprüfung stellt keine Transaktionen oder genau einmal ausgeführte Vorgänge für Agent- oder Tool-Nebeneffekte bereit, sodass Anwendungen weiterhin überlappende Anforderungen koordinieren müssen.

Übergeben Sie für antwortspezifischen Speicher einen StoreProvider an einen function_approval_store_provider oder einen ContextScopedStoreProvider an einen checkpoint_store_provider.

Äußere Hintergrundaufgaben verwenden das für den Aufrufer sichtbare response.id zum Abfragen. background_source="agent_server" Standardmäßig wird die Hintergrundausführung auf dem Host beibehalten. Legen Sie background_source="provider" nur mit history_source="service" und einem speichernden, fortsetzbaren Responses-Client fest. Wenn ResponsesServerOptions(resilient_background=True) ebenfalls festgelegt ist, kann der Host die Providerabfrage erst fortsetzen, nachdem er den privaten Fortsetzungstoken gespeichert hat. Machen Sie lokale Tool-Seiteneffekte idempotent, da ein Absturz, bevor das nächste Token gespeichert ist, sie erneut auslösen kann.

Importieren Sie azure.ai.agentserver.responses aus ResponsesServerOptions und übergeben Sie es über den Parameter options an ResponsesHostServer. Die verfügbaren Optionen für lange Unterhaltungen hängen vom Agenttyp ab:

Vollständige Implementierungen finden Sie unter dem benutzerdefinierten Speicher, dem grundlegenden Antwortverlauf und dem Hintergrund sowie stabilen, dauerhaft ausgeführten Workflowbeispielen .

Lesen von Dateien aus der gehosteten Sandbox

Behandeln Sie die persistente $HOME Ressource eines gehosteten Sandkastens als Anforderungsroutenressource, nicht als allgemeine Dateisystemgrenze. Akzeptieren Sie nur Dateien, die Ihre Anwendung explizit in ein dediziertes Verzeichnis hochlädt, überprüfen Sie die aktuelle Sandkastenidentität, und lehnen Sie absolute Pfade, Durchlaufen, Links, nichtregulare Dateien und übergroße oder ungültige Inhalte ab.

Leiten Sie für das Responses-Protokoll eine Anfrage mit dem Feld agent_session_id body an eine gehostete Sitzung weiter. Der Query-String-Selektor ist für die Aufrufe. Sitzungsuploads und Toolbox-Codedolmetscherdateien sind separate Ressourcen; Eine hochgeladene Sandkastendatei wird nicht automatisch in einen Toolboxcontainer eingebunden. Sehen Sie sich das Beispiel für Sitzungsdateien für gebundene UTF-8-Lesevorgänge und lokale und gehostete Uploadanleitungen an.

Anforderungsoptionen steuern

Der Host ordnet native Antwortgenerierungsfelder den Ausführungsoptionen des Agent-Frameworks zu. Beispielsweise wird max_output_tokens zu max_tokens, und parallel_tool_calls wird zu allow_multiple_tool_calls. Abgeflachte Werte aus extra_body überschreiben übersetzte native Werte.

Verwenden Sie den synchronen oder asynchronen prepare_options(request, options) Hook, um Anrufermodelloptionen zu entfernen oder zu ersetzen, bevor ein normaler Agent ausgeführt wird. Der Hook kann keine hostgesteuerte Identitäts-, Speicher-, Fortsetzungs- oder Transportfelder festlegen. Für eine benutzerdefinierte SupportsAgentRun-Implementierung, die keine Modelloptionen zur Laufzeit akzeptieren kann, legen Sie "warn" auf unsupported_options (Standardeinstellung), "ignore" oder "error" fest.

Behandeln von OAuth-Zustimmungsanforderungen

Wenn ein von Foundry gehostetes MCP-Tool die Zustimmung des Benutzers erfordert, gibt ResponsesHostServer eine unvollständige Antwort mit einem oauth_consent_request-Ausgabeelement zurück. Präsentieren Sie dem Benutzer consent_link, und fahren Sie dann mit der ID der unvollständigen Antwort als previous_response_id fort, sobald der Benutzer seine Zustimmung erteilt hat. Der Host behält die Agent-Sitzung für diesen erneuten Versuch bei und stellt ausschließlich absolute HTTPS-Einwilligungslinks bereit.

Wenn Ihr Host die erwarteten Autorisierungsursprünge kennt, beschränken Sie Einwilligungslinks mit allowed_oauth_consent_origins:

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

Beim Weglassen der Zulassungsliste wird die absolute HTTPS-Überprüfung beibehalten, ohne den Zielursprung einzuschränken. Wenn Sie eine leere Liste angeben, werden alle Zustimmungslinks abgelehnt. Konfigurieren Sie nur genaue HTTPS-Ursprünge; Einträge mit einem Pfad, einer Abfrage oder einem Fragment werden abgelehnt.

Mit dem Invocations-Protokoll in C# implementieren Sie einen benutzerdefinierten InvocationHandler, um eingehende Anforderungen zu verarbeiten:

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

Die AddInvocationsServer Methode registriert die Invocations-Protokolldienste. Sie implementieren InvocationHandler , um zu definieren, wie Ihr Agent jede Anforderung verarbeitet.

Verwenden Sie InvocationsHostServer für ein einfaches Setup aus dem agent_framework_foundry_hosting Paket. Es umschließt Ihren Agenten ähnlich wie ResponsesHostServer und behandelt die Sitzungsverwaltung automatisch.

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 akzeptiert dieselben Instanz- oder anforderungsgebundenen Factory-Formen, die für den Responses-Host beschrieben werden. Es stellt serialisierte Sitzungen aus dem konfigurierten Speicher wieder her, sodass abgeschlossene Unterhaltungen nach dem Neustart des Hosts fortgesetzt werden können. Informationen zu Speicherverhalten, Aufbewahrung und Anpassung finden Sie unter Status beibehalten und lang andauernde Konversationen verarbeiten.

Wenn gehostet, verwendet Aufrufe den überprüften Anforderungsbereich, der im Status "Persist" beschrieben ist, und behandelt lange ausgeführte Unterhaltungen. Behandeln Sie AgentSession.session_id als einen undurchsichtigen Wert; analysieren Sie ihn nicht und verlassen Sie sich nicht auf seine interne Darstellung. Lokale Ausführungen behalten das vorhandene Speicherverhalten von Einzelbenutzern bei.

Einen nativen Workflow mit Invocations hosten

Übergeben Sie workflow= und einen expliziten parse_request Callback, um einen systemeigenen Workflow zu hosten. Der Callback besitzt das JSON-Schema der Anwendung und gibt entweder eine WorkflowTurn mit typisierter Eingabe oder den vollständigen ausstehenden Antwortbatch zurück:

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

Fügen Sie jeden benutzerdefinierten Anwendungstyp hinzu, den der Workflow in der Liste des Checkpoint-Providers allowed_checkpoint_types speichert.

Gehostete Workflows erfordern eine anfragekontextsensitive Factory, die einen neu erstellten Graphen mit stabilen Workflow- und Executor-IDs zurückgibt. Ein direkt erstellter Workflow ist nur für eine lokale, einmalige Ausführung verfügbar, die nicht angehalten wird.

Nicht-Streaming-Workflowantworten verwenden Anwendungs-JSON mit einer output Ereignisliste. Beim Streaming werden framed output- und request_info-Ereignisse ausgegeben, gefolgt von done erst, nachdem der genaue Workflow-Cursor gespeichert wurde. Behandeln Sie die Streaming-Ausgabe bis done als vorläufig. legacy_wire_format=True wird von nativen Workflows nicht unterstützt.

Der Host überprüft Antworten anhand des genau passenden ausstehenden Prüfpunkts im vertrauenswürdigen Benutzer- und Sandkastenbereich. Wenn ein Workflow über mehrere ausstehende Anfragen verfügt, antworten Sie in einem Durchgang auf den vollständigen Batch. Einen ausführbaren Parser, einen typisierten Ticket-Workflow, eine Zulassungsliste für Prüfpunkttypen und JSON/SSE-Beispiele finden Sie im Workflowbeispiel für systemeigene Aufrufe.

Anpassen von Aufrufanforderungen und Antworten

Standardmäßig akzeptiert POST /invocations ein JSON-Objekt mit einer Zeichenfolge message, einem optionalen options-Objekt und einem optionalen booleschen Wert stream. Um eine anwendungsspezifische Nutzlast zu akzeptieren, übergeben Sie einen synchronen oder asynchronen parse_request Callback, der InvocationRun(messages, options, stream) zurückgibt. Verwenden Sie prepare_options, um eine Kopie der Aufrufergenerierungsoptionen zu filtern oder zu ersetzen, bevor der Agent ausgeführt wird.

Der Host validiert die Ausgabe des Hooks und weist Kontrollen für Plattformidentität, Speicherung, Fortsetzung und Agentenausführung zurück. Für Agenten, die keine Laufzeitoptionen akzeptieren, setzen Sie unsupported_options auf "warn" (Standard), "ignore" oder "error". Eine vollständige Implementierung finden Sie im Beispiel „Invocations-Parser“.

Bei nicht gestreamten erfolgreichen Antworten wird JSON in der Form {"response": "..."} zurückgegeben. Streaming verwendet vom Server gesendete Ereignisse: ein oder mehrere event: delta-Frames, gefolgt von event: done bei Erfolg oder event: error bei Fehlschlag. Ein Stream kann vor einem Fehler Deltas ausgeben, daher müssen Clients done, nicht ein Delta, als erfolgreichen Abschluss behandeln. Der Host gibt done erst aus, nachdem er den Antwortstream abgeschlossen und AgentSession dauerhaft gespeichert hat. Dies session_id ist die Plattform-Sandkastenrouten-ID, nicht die serialisierte AgentSession.session_id.

Setzen Sie legacy_wire_format=True nur während der Migration vorhandener Clients, die die bisherige Klartextantwort und den Rohdatenstrom aus Textsegmenten benötigen. Dieser Kompatibilitätsmodus ist veraltet und konvertiert keine Fehler in erfolgreichen Text. Der Host serialisiert Anfragen aus derselben Sitzung nur innerhalb eines Prozesses; ein prozessübergreifender Compare-and-Swap-Konflikt kann nach Auswirkungen externer Tools weiterhin auftreten.

Das Aufrufprotokoll setzt keine Workflowausführung fort, die aussteht oder unterbrochen wird. Verwenden Sie das benutzerdefinierte Handlermuster im folgenden Abschnitt, wenn Sie ein anderes Verhalten für die Workflowfortsetzung benötigen.

Um die vollständige Kontrolle über die Anforderungsverarbeitung zu erhalten, verwenden Sie InvocationAgentServerHost direkt aus dem Paket azure.ai.agentserver.invocations, und implementieren Sie Ihren eigenen Aufrufhandler:

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

Eine vollständige Bereitstellung für Invocations finden Sie im von Foundry gehosteten Telegram-Beispiel. Es platziert API Management vor den gehosteten Agent-Webhooks und verwendet verwaltete Identitäten, Key Vault und Cosmos DB für einen persistenten Gesprächsverlauf.