Agent Server

Ein Agent-Server ist die Bibliothek, die deinen Agent-Code in einen Service umwandelt. Er kapselt die Agentenschleife in einen HTTP-Server, definiert die API, die Clients aufrufen, um den Agenten auszuführen, verwaltet die Client-Verbindungen und bestimmt, was passiert, wenn ein Run unterbrochen wird. Der Agent-Server läuft auf der Agent-Runtime. Um zu erfahren, wie die Schichten zusammenpassen, siehe Agents auf Azure Databricks bereitstellen.

Für neue Agenten empfiehlt Databricks den von AgentKit bereitgestellten Agent-Server, der auf dieser Seite beschrieben wird. Um es mit den Legacy-Agent-Servern zu vergleichen, siehe Vergleichen von Agentenservern auf Azure Databricks.

Anforderungen

• Python 3.10 und höher.
• Das Paket databricks-agentbricks beinhaltet AgentKit. Projekte, die Sie mit agentbricks init erstellen, deklarieren es als Abhängigkeit.

AgentKit

AgentKit ist die Agent Bricks-Bibliothek zum Bau von Agenten, die auf Agent Runtime laufen. Es stellt den Agentenserver bereit, der Ihren Agenten bedient, sowie Clients und Hilfsprogramme für Agentenspeicher, Agentensitzungen, Nachverfolgung und Authentifizierung. Projekte, die du mit der Agent Bricks CLI erstellst, verwenden standardmäßig AgentKit.

Agent-Server

AgentKit stellt seinen Agent-Server über die DurableAgentServer-Klasse zur Verfügung. Der Agent-Server bettet deine Agent-Schleife in einen HTTP-Server ein, der die Invocation-API bereitstellt, jeden Lauf protokolliert und durch einen Absturz oder Neustart unterbrochene Läufe fortsetzt. Sie bietet:

• Eine API für jeden Anforderungsmodus: Synchrone, Streaming- und Hintergrundaufrufe sowie Stream-Wiederverbindung, alle vom selben Handler bedient.
• Idempotente Aufrufe: Eine vom Client generierte Aufruf-ID stellt sicher, dass eine erneute Anfrage keinen doppelten Lauf startet.
• Sequenzierte Sitzungen: Aufrufe in derselben Sitzung werden nacheinander, jeweils nur eine auf einmal, in der richtigen Reihenfolge ausgeführt.
• Persistenter Laufzustand: Bei der Bereitstellung bleiben Laufstatus, Ereignisse und Ergebnisse nach Worker-Neustarts erhalten.
• Absturz-Wiederherstellung: Der Server erkennt unterbrochene Läufe und startet einen erneuten Versuch.
• Autorisierung des anfordernden Benutzers: Tools können mit den Berechtigungen des Benutzers agieren, der die Anforderung gesendet hat.
• Benutzerdefinierte Endpunkte: Der Agent-Server ist eine FastAPI-Anwendung, sodass du deine eigenen Routen hinzufügen kannst.

Was AgentKit umfasst

AgentKit ist die databricks_agentkit Python-Bibliothek. Es exportiert Folgendes:

Die Bibliothek enthält außerdem Framework-Helfer in databricks_agentkit.langgraph und databricks_agentkit.openai, die die generierten Vorlagen verwenden, um jedes Framework mit dem Session Store zu verbinden. Für die Speicher- und Sitzungs-APIs siehe Agent-Speicher und Agent-Sitzungen.

Registrieren Sie Ihren Agenten

Wenn Sie ein Projekt mit agentbricks initerstellen, übernimmt die CLI dies für Sie. Der generierte runtime/main.py erstellt den Server und registriert die Aufruf- und Wiederherstellungshandler der Vorlage, sodass man nur den Agentencode in agent/ bearbeitet. Führen Sie die Schritte in diesem Abschnitt aus, um einen vorhandenen Agent einzubringen oder einen eigenen Handler zu schreiben.

Erstelle ein DurableAgentServer und registriere einen asynchronen Invocation-Handler mit @app.invoke. Der Handler erhält das input der Anfrage und einen Aufrufkontext und liefert ein JSON-serialisierbares Ergebnis. Veröffentlichen Sie den Fortschritt als Ereignisse mit context.emit.

from databricks_agentkit import DurableAgentServer, InvocationContext

app = DurableAgentServer()


@app.invoke
async def invoke(input, context: InvocationContext) -> dict:
    await context.emit({"type": "status", "message": "Looking that up"})
    answer = await run_my_agent(input, session_id=context.session_id)
    return {"answer": answer}

Du kannst einen Invocation-Handler registrieren, und der Server startet nicht ohne einen. Der Handler bedient jeden Anforderungsmodus: Der Client wählt aus, ob er auf das Ergebnis warten, Ereignisse streamen oder im Hintergrund ausgeführt werden soll.

Um den Server lokal zu betreiben, starte ihn mit agentbricks dev. Projekte, die du mit agentbricks init erstellst, beinhalten einen Entrypoint, der den Server mit Uvicorn betreibt, und eine app.yaml Datei, die denselben Entrypoint nach dem Deployment startet.

Aufrufkontext

Das zweite Argument des Handlers ist ein InvocationContext:

Invocation-API

DurableAgentServer bedient die Aufruf-API unter /api/invocations:

• POST /api/invocations startet einen Aufruf. Standardmäßig wartet die Anfrage auf das Ergebnis. Legen Sie stream fest, um Ereignisse als Server-Sent Events zu empfangen, oder background, um sofort mit einer Status-URL zurückzukehren.
• GET /api/invocations/<id> gibt den Status eines Aufrufs zurück und nach Abschluss dessen Ausgabe.
• Mit GET /api/invocations/<id>/events?after=<event-id> werden gespeicherte Ereignisse gestreamt, sodass ein Client sich nach einer abgebrochenen Verbindung wieder verbinden kann.

Anforderungsfelder, Beispiele und Antwortformate finden Sie unter Abfragen von in Azure Databricks bereitgestellten Agents.

Idempotenz

Clients senden bei jedem Aufruf eine UUID id. Der Server behandelt die ID als Idempotenzschlüssel, während er den Aufruf-Datensatz beibehält: Das erneute Senden derselben Anfrage bringt den bestehenden Aufruf zurück, anstatt den Agenten erneut auszuführen. Die Wiederverwendung einer ID für eine andere Anfrage liefert einen 409 Fehler.

Sitzungen

Clients können eine session_id senden, um Aufrufe in einer Unterhaltung zu gruppieren. Der Server speichert die Sitzungs-ID separat von input, gibt sie an deinen Handler als context.session_id und führt Aufrufe, die jeweils eine Sitzungs-ID teilen, jeweils nacheinander in der angegebenen Reihenfolge aus. Der Server leitet keine Sitzung aus der Aufruf-ID oder dem Input ab. Ohne Sitzungs-ID erfolgt ein Aufruf ohne Sitzung.

Ausführungsstatus

DurableAgentServer speichert die Anforderung, den Status, die Heartbeats, die Ereignisse und das Ergebnis jedes Aufrufs in einem Runtime Store.

• Lokale Entwicklung: agentbricks dev verwendet einen In-Process-Runtime Store. Die Aufruf-API verhält sich genauso, aber der Laufzustand geht verloren, wenn der Prozess stoppt und der Server die unterbrochene Arbeit nicht neu startet.
• Bereitgestellte Agents: agentbricks deploy stellt eine dedizierte Datenbank für den Runtime Store jeder Bereitstellung in einem von Azure Databricks verwalteten Lakebase-Projekt bereit und verwendet sie bei einer erneuten Bereitstellung wieder. Du kannst dein eigenes Lakebase-Projekt nicht für den Runtime Store verwenden und erstellst oder bindest es nicht selbst. Ergebnisse und Ereignisse überdauern Worker-Neustarts, und jede Instanz des Agenten kann Status- und Wiederverbindungsanfragen bearbeiten. agentbricks deployments delete entfernt den Runtime Store mit der Bereitstellung.

Der Runtime Store enthält den Ausführungszustand des Servers. Er ist von den Sitzungs- und Memory-Stores getrennt, die Ihr Agent für den Unterhaltungsverlauf und den Langzeitspeicher verwendet.

Wiederherstellung nach Absturz

Um Ausführungen wiederherzustellen, die durch einen Worker-Absturz oder -Neustart unterbrochen wurden, registrieren Sie einen Wiederherstellungshandler bei @app.recover. Wenn ein eingesetzter Server erkennt, dass die Heartbeats eines Runs ausgeblieben sind, startet er einen Ersatzversuch auf einem verfügbaren Worker und ruft den Wiederherstellungshandler mit der ursprünglichen Eingabe auf.

@app.recover
async def recover(input, context: InvocationContext) -> dict:
    # Resume from the agent's last checkpoint in the session store,
    # or replay the input if that's safe for your agent.
    return await resume_my_agent(input, session_id=context.session_id)

Wenn du keinen Wiederherstellungshandler registrierst, wird die automatische Wiederherstellung deaktiviert und der Server meldet eine Warnung, wenn er startet.

Die Wiederherstellung funktioniert wie folgt:

• Wenn die Wiederherstellung beginnt: Jeder Laufversuch sendet alle paar Sekunden einen Heartbeat. Wenn die Heartbeats stoppen, beispielsweise weil der Worker abstürzt, neu startet oder während einer erneuten Bereitstellung ersetzt wird, erkennt der Server die veraltete Ausführung innerhalb von Sekunden und startet einen Ersetzungsversuch.
• Wenn die Wiederherstellung nicht gestartet wird: Wenn Ihr Handler eine Ausnahme auslöst, schlägt der Aufruf fehl, und der Server versucht es nicht erneut. Die Wiederherstellung deckt unterbrochene Worker ab, keine Fehler in Ihrem Agent-Code.
• Anzahl der Versuche: Der Server begrenzt die Anzahl der Wiederherstellungsversuche nicht. Jeder Ersatzversuch erhöht context.attempt um eins. Um nach einer bestimmten Anzahl von Versuchen zu stoppen, überprüfen Sie context.attempt in Ihrem Wiederherstellungshandler, und lösen Sie einen Fehler aus.
• Manuelle Wiederherstellung: Du kannst die Wiederherstellung nicht manuell auslösen. Das erneute Senden einer Anfrage mit derselben Aufruf-ID gibt den bestehenden Aufruf zurück, anstatt einen neuen Versuch zu starten.

Recovery kann deinen Agentencode mehrmals für denselben Aufruf ausführen. Ein unterbrochener Versuch könnte bereits externe Systeme aufgerufen haben, bevor der Ersatzversuch beginnt, also machen Sie diese Aufrufe idempotent.

Autorisierung des anfordernden Benutzers

Standardmäßig laufen die Tools deines Agenten mit den Berechtigungen des Service Principal der App. Um ein Tool mit den Berechtigungen des Benutzers auszuführen, der die Anfrage gesendet hat, deklarieren Sie die Benutzerautorisierung in agent.toml:

• Legen Sie für ein verwaltetes Tool auth = "user" im Tooleintrag fest. Die agentbricks tools add Befehle für MCP-Server, Sandboxen und Genie Agents schreiben auth = "user" standardmäßig. Übergeben Sie --auth app, um stattdessen die Identität der App zu verwenden.

• Für ein Tool, das du im Code schreibst, deklariere die Anforderung und alle API-Scopes, die Agent Bricks nicht ableiten kann:

  [auth.user]
  required = true
  additional_api_scopes = ["sql"]
  

Wenn ein Agent eine Benutzerautorisierung benötigt, DurableAgentServer liest er die Zugangsdaten des Benutzers aus den vertrauenswürdigen Databricks Apps Request-Headern und speichert sie nur für den aktiven Versuch im Speicher. Der Runtime Store speichert die Zugangsdaten nicht. In Ihrem Handler erhalten Sie einen Workspace-Client für den Benutzer von context.request_auth:

@app.invoke
async def invoke(input, context: InvocationContext) -> dict:
    user_client = context.request_auth.client_for("user")
    me = user_client.current_user.me()
    return {"answer": f"Hello, {me.user_name}"}

client_for("app") gibt einen Client zurück, der den Service-Principal der App nutzt. Der Resolver schließt, wenn der Versuch endet, also rufe ihn im Handler auf, anstatt den Client zu speichern. Wenn du den Agent lokal mit agentbricks dev ausführst, verwendet client_for("user") deine lokalen Zugangsdaten.

Wenn Sie deployen, agentbricks deploy fordern Sie die Benutzerscopes von Databricks Apps an, die Ihre Tools benötigen. Um fehlende Scopes zu einer bestehenden App hinzuzufügen, übergeben Sie --allow-user-scope-update. Siehe Konfigurieren der Autorisierung in einer Databricks-App.

Aufrufe des anfordernden Benutzers verwenden dieselben synchronen, Streaming-, Hintergrund- und Wiederverbindungs-APIs. Da der Server die Anmeldeinformationen des Benutzers nicht speichert, kann er einen unterbrochenen Aufruf des anfordernden Benutzers nicht wiederherstellen. Der Ersetzungsversuch schlägt mit dem Fehler MCP_USER_AUTH_RECOVERY_UNSUPPORTED fehl, bevor Ihre Handler ausgeführt werden.

Benutzerdefinierte Endpunkte hinzufügen

DurableAgentServer ist eine FastAPI-Anwendung. Fügen Sie Routen neben der Aufruf-API auf die gleiche Weise hinzu, wie Sie sie zu jeder FastAPI-App hinzufügen:

@app.get("/status")
async def status() -> dict:
    return {"ready": True}

Rahmenvorlagen

agentbricks init erzeugt zwei Verzeichnisse:

• agent/ enthält deinen Framework-Code: das Modell, Prompts und Tools.
• runtime/ enthält den Adapter, der das Framework mit DurableAgentSerververbindet, sowie den Entrypoint, der die Aufruf- und Wiederherstellungshandler des Adapters registriert.

Der Adapter übersetzt jeden Aufruf in einen Aufruf zur Agentenschleife des Frameworks und übersetzt die Ausgabe des Frameworks in Ereignisse und ein Ergebnis. Beide Vorlagen registrieren einen Recovery-Handler. Die LangGraph-Vorlage wird ab dem letzten Checkpoint im Sitzungsspeicher fortgesetzt, und die OpenAI Agents SDK-Vorlage spielt die Anfrage in derselben Sitzung erneut ab. Um einen vorhandenen Agenten einzubinden, fügen Sie einen Adapter und einen DurableAgentServer Entrypoint hinzu und setzen Sie server = "agentbricks" im Abschnitt [agent] von agent.toml.

Agentserver auf Azure Databricks vergleichen

Azure Databricks stellt drei Agent-Server bereit: den AgentKit-Agentenserver und zwei Legacy-Agent-Server, die von früheren App-Vorlagen verwendet werden.

LongRunningAgentServer erweitert das MLflow AgentServer, und beide bedienen Agenten, die die MLflow-Schnittstelle ResponsesAgent implementieren. Informationen zum Bereitstellen und Verwalten eines Agents, der einen davon verwendet, finden Sie unter Ausführen von Agents in Databricks-Apps mithilfe des Legacy-Agent-Servers. Um einen Agenten auf einem dieser Server abzufragen, siehe Auf Azure Databricks bereitgestellte Agents abfragen.

Limitations

• Du kannst den Agentenserver einer bestehenden Bereitstellung nicht ändern. Um zwischen DurableAgentServer und Ihrem eigenen Server zu wechseln, erstellen Sie ein neues Projekt mit der gewünschten Option agentbricks init --server, und stellen Sie es unter einem neuen Namen bereit.
• Das Ändern des Feldes server in agent.toml konvertiert den vorhandenen Servercode nicht zu DurableAgentServer.
• Das Anfordern der Benutzerautorisierung erfordert server = "agentbricks".

Weitere Ressourcen

• Agent-Runtime
• Auf Azure Databricks bereitgestellte Abfrage-Agenten
• Agent Bricks CLI