Agent Server

Un agent server è la libreria che trasforma il codice del tuo agente in un servizio. Racchiude il ciclo dell'agente in un server HTTP, definisce l'API che i client chiamano per eseguire l'agente, gestisce le connessioni dei client e determina cosa succede quando una run viene interrotta. Il server dell'agente viene eseguito sul runtime dell'agente. Per capire come i livelli si incastrano, vedi Deploy agents su Azure Databricks.

Agent servers on Azure Databricks

Azure Databricks fornisce tre server agent. Per i nuovi agenti, Databricks raccomanda DurableAgentServer.

Server dell'agente Pacchetto API Client Esecuzione durevole Usato da
DurableAgentServer (scelta consigliata) databricks_agentkit, nel databricks-agentbricks pacchetto API di invocazione a /api/invocations: esecuzioni sincrone, streaming e in background, con riconnessione del flusso Uno Store Runtime che agentbricks deploy effettua il provisioning, più il recupero in caso di crash tramite un gestore di recupero Progetti che crei con la CLI di Agent Bricks
LongRunningAgentServer (eredità) databricks_ai_bridge.long_running, nel databricks-ai-bridge[agent-server] pacchetto API OpenAI Responses a /responses, con esecuzioni in background e ripresa del flusso Stato di esecuzione in un database Lakebase che configuri. Dopo un crash, un nuovo tentativo continua l'esecuzione dal registro eventi del tentativo interrotto. Il e agent-openai-advancedagent-langgraph-advanced
MLflow AgentServer (legacy) mlflow.genai.agent_server, nel mlflow pacchetto API OpenAI Responses a /responses: esecuzioni sincrone e streaming None I modelli base delle app, come agent-openai-agents-sdk

LongRunningAgentServer estende il MLflow AgentServer, e entrambi servono agenti che implementano l'interfaccia MLflow ResponsesAgent . Per distribuire e mantenere un agente che ne utilizza uno di essi, vedi Esegui agenti su Databricks Apps usando il server agente legacy. Per interrogare un agente su uno qualsiasi di questi server, vedi Eseguire query sugli agenti distribuiti in Azure Databricks.

DurableAgentServer

DurableAgentServer è l'agente server Agent Bricks. Avvolge il loop dell'agente in un server HTTP che serve l'API di invocazione, traccia ogni esecuzione e ripristina le run interrotte da un crash o da un riavvio. Gli agenti che crei con la CLI Agent Bricks usano DurableAgentServer di default.

DurableAgentServer offre:

  • Un'API per ogni modalità di richiesta: sincrona, streaming e invocazioni in background, più riconnessione dello stream, tutte servite dallo stesso handler.
  • Invocazioni idempotenti: Un ID di invocazione generato dal client garantisce che una richiesta riprovata non avvii un'esecuzione duplicata.
  • Sessioni ordinate: Le invocazioni nella stessa sessione vengono eseguite una alla volta, in ordine.
  • Stato di esecuzione persistente: Quando viene distribuito, lo stato dell'esecuzione, gli eventi e i risultati sopravvivono ai riavvii dei lavoratori.
  • Recupero in caso di crash: Il server rileva le esecuzioni interrotte e avvia un tentativo di sostituzione.
  • Richiesta con autorizzazione utente: Gli strumenti possono agire con i permessi dell'utente che ha inviato la richiesta.
  • Endpoint personalizzati: DurableAgentServer è un'applicazione FastAPI, quindi puoi aggiungere le tue rotte.

Requirements

DurableAgentServer ha i seguenti requisiti:

  • Python 3.10 e superiori.
  • Il databricks-agentbricks pacchetto, che include la databricks_agentkit libreria. I progetti che crei con agentbricks init lo dichiarano come dipendenza.

Registra il tuo agente

Quando crei un progetto con agentbricks init, la CLI lo fa per te. Il file generato runtime/main.py crea il server e registra i gestori di invocazione e ripristino del modello, quindi modifichi solo il codice dell'agente in agent/. Segui i passaggi in questa sezione per integrare un agente esistente o per scrivere il tuo handler.

Crea un DurableAgentServer e registra un gestore di invocazione asincrono con @app.invoke. Il gestore riceve il/la input della richiesta e un contesto di invocazione, e restituisce un risultato serializzabile in JSON. Pubblica l'avanzamento come eventi con 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}

Puoi registrare un solo gestore di invocazioni, e il server non parte senza di uno. Il gestore gestisce tutte le modalità di richiesta: il client sceglie se aspettare il risultato, trasmettere gli eventi in streaming o eseguire in background.

Per far girare il server localmente, avvialo con agentbricks dev. I progetti che crei con agentbricks init includono un punto d'ingresso che esegue il server con Uvicorn e un file app.yaml che avvia lo stesso punto di ingresso dopo il deployment.

Contesto di chiamata

Il secondo argomento dell'handler è un InvocationContext:

Attribute Description
invocation_id L'ID che il cliente ha inviato per questa invocazione.
session_id La sessione a cui appartiene l'invocazione, oppure None se il client non ne ha inviata una.
attempt Il numero del tentativo. Il primo tentativo è 1.
is_recovery True quando il responsabile del recupero sta eseguendo un tentativo di sostituzione.
emit(event) Memorizza un evento JSON, lo consegna ai client di streaming e restituisce la posizione dell'evento nel flusso.
request_auth Il resolver delle credenziali request-user, quando l'agente richiede l'autorizzazione request-user. In caso contrario, None.

API di invocazione

DurableAgentServer serve l'API di invocazione a /api/invocations:

  • POST /api/invocations inizia un'invocazione. Per impostazione predefinita, la richiesta attende il risultato. Imposta stream per ricevere eventi come Server-Sent Eventi, oppure background per restituire immediatamente un URL di stato.
  • GET /api/invocations/<id> restituisce lo stato di un'invocazione e, dopo il completamento, il suo output.
  • GET /api/invocations/<id>/events?after=<event-id> Streama gli eventi memorizzati, così un client può riconnettersi dopo una connessione interruta.

Per campi di richiesta, esempi e formati di risposta, vedi Query agents deployed on Azure Databricks.

Idempotenza

I clienti inviano un UUID id con ogni invocazione. Il server tratta l'ID come una chiave di idempotenza mentre mantiene il record di invocazione: riinviare la stessa richiesta restituisce l'invocazione esistente invece di eseguire di nuovo l'agente. Riutilizzare un ID per una richiesta diversa restituisce un errore 409.

Sessioni

I client possono inviare un session_id per raggruppare le invocazioni in una sola conversazione. Il server memorizza l'ID della sessione separatamente da input, lo passa al tuo gestore come context.session_id, ed esegue invocazioni che condividono un ID sessione una alla volta, in ordine. Il server non deduce una sessione dall'ID dell'invocazione o dall'input. Senza un ID sessione, un'invocazione è senza sessione.

Stato di esecuzione

DurableAgentServer memorizza la richiesta, lo stato, i battiti cardiaci, gli eventi di ogni invocazione e il risultato in uno Store di Runtime.

  • Sviluppo locale: agentbricks dev utilizza un Runtime Store nel processo. L'API di invocazione si comporta allo stesso modo, ma lo stato di esecuzione si perde quando il processo si ferma e il server non riavvia il lavoro interrotto.
  • Agenti distribuiti nell'ambiente di runtime: agentbricks deploy fornisce un database dedicato per il Runtime Store di ogni distribuzione in un progetto Lakebase gestito da Azure Databricks e lo riutilizza quando si ridistribuisce. Non puoi usare il tuo progetto Lakebase per il Runtime Store, e non lo crei né lo associ da solo. I risultati e gli eventi sopravvivono ai riavvii dei lavoratori, e qualsiasi istanza dell'agente può gestire le richieste di stato e di riconnessione. agentbricks deployments delete rimuove lo Runtime Store con il deployment.

Lo Store Runtime contiene lo stato di esecuzione del server. È separato dagli archivi di sessione e memoria che il tuo agente usa per la cronologia delle conversazioni e la memoria a lungo termine.

Recupero in caso di crash

Per recuperare le run che l'arresto anomalo o il riavvio di un worker interrompe, registra un gestore di recupero con @app.recover. Quando un server deployato rileva che i battiti cardiaci di una run si sono fermati, avvia un tentativo di sostituzione su un worker disponibile e chiama il gestore di recupero con l'input originale.

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

Se non registri un gestore di recupero, il recupero automatico viene disattivato e il server scrive un avviso nel log all'avvio.

Il recupero funziona come segue:

  • Quando inizia il recupero: ogni tentativo in esecuzione invia un heartbeat ogni pochi secondi. Se gli heartbeat si fermano, ad esempio perché il worker crasha, si riavvia o viene sostituito durante una ridistribuzione, il server rileva l'esecuzione obsoleta in pochi secondi e avvia un tentativo di esecuzione sostitutiva.
  • Quando il recovery non inizia: se il tuo handler solleva un'eccezione, l'invocazione fallisce e il server non la riprova. Il recupero copre i worker interrotti, non gli errori nel codice del tuo agente.
  • Numero di tentativi: Il server non limita il numero di tentativi di recupero. Ogni tentativo di sostituzione incrementa context.attempt di 1. Per interrompere dopo un certo numero di tentativi, controlla context.attempt nel tuo gestore di recupero e segnala un errore.
  • Recupero manuale: Non puoi attivare il recupero manualmente. Riinviare una richiesta con lo stesso ID di invocazione restituisce l'invocazione esistente invece di iniziare un nuovo tentativo.

Il recovery può eseguire il codice dell'agente più di una volta per la stessa invocazione. Un tentativo interrotto potrebbe aver già chiamato sistemi esterni prima che iniziasse il tentativo di sostituzione, quindi assicurati che quelle chiamate siano idempotenti.

libreria AgentKit

DurableAgentServer fa parte della libreria AgentKit, databricks_agentkitche il databricks-agentbricks pacchetto include. Progetti che crei con agentbricks init importato da esso. La libreria esporta le seguenti funzioni di supporto:

Esporta Description
DurableAgentServer, InvocationContext Il server agente e il contesto che esso passa ai tuoi gestori di invocazione e ripristino.
AgentKitClient Un client per la memoria gestita e archivi di sessione. Crea e ottiene store, ed espone le memorie e le sessioni degli store come oggetti Memory, MemoryStore, MemorySearchResult, Session, SessionStore e SessionItem.
configure_tracing, start_trace Imposta il tracciamento MLflow per l'agente e avvia un tracciamento attorno a un'unità di lavoro.
workspace_client, workspace_headers Crea un SDK WorkspaceClientDatabricks autenticato, oppure ottieni header di autenticazione per chiamate HTTP dirette, dall'ambiente dell'agente.
list_ai_gateway_model_services Elenca i servizi modello che l'agente può chiamare tramite Unity Gateway.

La libreria include anche helper del framework in databricks_agentkit.langgraph e databricks_agentkit.openai, che i template generati usano per collegare ogni framework allo store delle sessioni. Per le API di memoria e sessione, vedi Memoria di agente gestita e Sessioni di agente gestito.

Autorizzazione utente richiesta

Per impostazione predefinita, gli strumenti del tuo agente funzionano con le autorizzazioni del service principal dell'app. Per eseguire uno strumento con i permessi dell'utente che ha inviato la richiesta, dichiara l'autorizzazione utente in agent.toml:

  • Per uno strumento gestito, imposta auth = "user" sulla voce dello strumento. I comandi per server MCP, sandbox e Genie Agents scrivono auth = "user" di default. Specifica --auth app per usare invece l'identità dell'app.

  • Per uno strumento che scrivi in codice, dichiara il requisito e qualsiasi ambito API che Agent Bricks non può dedurre:

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

Quando un agente richiede l'autorizzazione dell'utente, DurableAgentServer legge la credenziale dell'utente dalle intestazioni di richiesta Databricks Apps affidabili e la conserva in memoria solo per il tentativo attivo. Il Runtime Store non memorizza la credenziale. Nel tuo handler, ottieni un client workspace per l'utente da: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") restituisce un client che utilizza il service principal dell'app. Il resolver si chiude quando il tentativo termina, quindi chiamalo all'interno del gestore invece di memorizzare il client. Quando esegui l'agente localmente con agentbricks dev, client_for("user") usa le tue credenziali locali.

Quando distribuisci, agentbricks deploy richiede gli scope utente di Databricks Apps di cui i tuoi strumenti hanno bisogno. Per aggiungere gli ambiti mancanti a un'app esistente, specifica --allow-user-scope-update. Consultare Configurare l'autorizzazione nell'applicazione Databricks.

Le invocazioni request-user utilizzano le stesse API sincrone, di streaming, di background e di riconnessione. Poiché il server non memorizza la credenziale dell'utente, non può recuperare un'invocazione richiesta-utente interrotta. Il tentativo di sostituzione fallisce con l'errore MCP_USER_AUTH_RECOVERY_UNSUPPORTED prima che i tuoi handler vengano eseguiti.

Aggiungi endpoint personalizzati

DurableAgentServer è un'applicazione FastAPI. Aggiungi le route insieme all'API di invocazione nello stesso modo in cui le aggiungi a qualsiasi app FastAPI:

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

Template del framework

agentbricks init genera due directory:

  • agent/ Contiene il tuo codice framework: il modello, i prompt e gli strumenti.
  • runtime/ contiene l'adattatore che collega il framework a DurableAgentServer, e il punto di ingresso che registra i gestori di invocazione e recupero dell'adattatore.

L'adattatore traduce ogni invocazione in una chiamata al ciclo agente del framework e traduce l'output del framework in eventi e un risultato. Entrambi i template registrano un gestore di recupero. Il template LangGraph riprende dal suo ultimo checkpoint nello store sessione, e il template SDK degli Agenti OpenAI riproduce la richiesta nella stessa sessione. Per integrare un agente esistente, aggiungi un adattatore e un punto di ingresso DurableAgentServer, e imposta server = "agentbricks" nella sezione [agent] di agent.toml.

Limitations

  • Non puoi cambiare il server agente di un deployment esistente. Per passare tra DurableAgentServer e il tuo server personale, crea un nuovo progetto con l'opzione agentbricks init --server che desideri e distribuiscilo con un nuovo nome.
  • Cambiare il server campo in agent.toml non converte il codice del server esistente in DurableAgentServer.
  • L'autorizzazione dell'utente richiedente richiede server = "agentbricks".

Risorse aggiuntive

  • Ultimo aggiornamento: