Agent Server
An agent server is the biblioteca that turns your agent code into a serviço. It wraps the agent loop in an HTTP server, defines the API that clients call to run the agent, gerencia client connections, and determines what happens when a execução is interrupted. The agent server runs on the agent runtime. To learn how the layers fit together, see Deploy agents on Databricks.
Servidores de agentes no Databricks
O Databricks fornece três servidores de agentes. Para novos agentes, o Databricks recomenda DurableAgentServer.
Servidor de agentes | Pacote | API do cliente | Execução durável | Usado por |
|---|---|---|---|---|
|
| API de invocação em | Um Armazenamento de Runtime que | Projetos que você cria com a CLI do Agent Bricks |
|
| OpenAI Responses API em | Execução state in a Lakebase database that you configure. After a crash, a new attempt continues the execução from the interrupted attempt's event log. | O |
MLflow |
| OpenAI Responses API em | Nenhuma | Os padrões de aplicativos base, como |
LongRunningAgentServer estende o MLflow AgentServer, e ambos atendem a agentes que implementam a interface MLflow ResponsesAgent. Para implantar e manter um agente que usa um deles, consulte Executar agentes no Databricks Apps usando o servidor de agente herddo. Para fazer uma query de um agente em qualquer um desses servidores, consulte Fazer query de agentes implantados no Databricks.
DurableAgentServer
DurableAgentServer is the Agent Bricks agent server. It wraps your agent loop in an HTTP server that serves the invocation API, tracks each execução, and recovers execuções that a crash or restart interrupts. Agents that you create with the Agent Bricks CLI use DurableAgentServer by default.
DurableAgentServer fornece:
- One API for every request mode : Synchronous, transmissão, and background invocations, plus transmissão reconnection, all served by the same handler.
- Invocações idempotentes : um ID de invocação gerado pelo cliente garante que uma solicitação retentada não comece uma execução duplicada.
- Sessões ordenadas : as invocações na mesma sessão entram em execução uma de cada vez, em ordem.
- Persistent run state : quando implantado, o status da execução, os eventos e os resultados sobrevivem às reinicializações do worker.
- Recuperação de falhas : o servidor detecta execuções interrompidas e começa uma tentativa de substituição.
- Autorização do usuário solicitante : as ferramentas podem agir com as permissões do usuário que enviou a solicitação.
- Endpoint personalizados :
DurableAgentServeré uma aplicação FastAPI, portanto você pode adicionar suas próprias rotas.
Requisitos
DurableAgentServer tem os seguintes requisitos:
- Python 3.10 and acima.
- O pacote
databricks-agentbricks, que inclui a bibliotecadatabricks_agentkit. Projetos criados comagentbricks inito declaram como uma dependência.
Registre seu agente
When you create a project with agentbricks init, a CLI faz isso para você. O runtime/main.py gerado cria o servidor e faz o registro dos manipuladores de invocação e recuperação do padrão, para que você edite apenas o código do agente em agent/. Siga os passos nesta seção para trazer um agente existente ou para escrever seu próprio manipulador.
Crie um DurableAgentServer e registro um manipulador de invocação assíncrona com @app.invoke. O manipulador recebe o input da solicitação e um contexto de invocação, e retorna um resultado serializável em JSON. Publique o progresso como eventos com 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}
Você pode fazer um registro de um manipulador de invocação e o servidor não começa sem um. O manipulador atende a todos os modos de solicitação: o cliente escolhe se deseja aguardar o resultado, fazer a transmissão de eventos, ou realizar a execução em segundo plano.
Para a execução do servidor localmente, comece-o com agentbricks dev. Projetos criados com agentbricks init incluem um ponto de entrada que executa o servidor com o Uvicorn e um arquivo app.yaml que começa o mesmo ponto de entrada após a implantação.
Contexto de execução
O segundo argumento do manipulador é um InvocationContext:
Atributo | Descrição |
|---|---|
| O ID que o cliente enviou para esta invocação. |
| The session that the invocation belongs to, or |
| The attempt number. The first attempt is |
|
|
| Armazena um evento JSON, o entrega a clientes de transmissão e retorna a posição do evento na transmissão. |
| O resolvedor de credenciais do usuário solicitante, quando o agente requer autorização do usuário solicitante. Caso contrário, |
API de invocações
DurableAgentServer fornece a API de invocação em /api/invocations:
POST /api/invocationscomeçar uma invocação. Por default, a solicitação aguarda o resultado. Definastreampara receber eventos como Server-Sent Events, oubackgroundpara retornar imediatamente com uma URL de status.GET /api/invocations/<id>retorna o status de uma invocação e, após a conclusão, sua saída.GET /api/invocations/<id>/events?after=<event-id>transmissões armazenados, para que um cliente possa se reconectar após uma conexão perdida.
Para campos de solicitação, exemplos e formatos de resposta, consulte Query agentes implantados no Databricks.
Idempotência
Os clientes enviam uma key UUID id a cada invocação. O servidor trata o ID como uma key de idempotência enquanto retiene o registro de invocação: o reenvio da mesma solicitação retorna a invocação existente em vez de executar o agente novamente. A reutilização de um ID para uma solicitação diferente retorna um erro 409.
Sessões
Os clientes podem enviar um session_id para agrupar invocações em uma única conversa. O servidor armazena o ID da sessão separadamente de input, o passa para o seu manipulador como context.session_id e executa invocações que compartilham um ID de sessão uma de cada vez, em ordem. O servidor não infere uma sessão a partir do ID de invocação ou da entrada. Sem um ID de sessão, uma invocação é sem sessão.
Estado de execução
DurableAgentServer stores each invocation's request, status, heartbeats, events, and result in a Runtime Store.
- Local development : o
agentbricks devusa um Runtime Store em processo. A API de invocação se comporta da mesma maneira, mas o estado de execução é perdido quando o processo é interrompido e o servidor não reinicia o trabalho interrompido. - Agentes implantados :
agentbricks deployprovisiona um banco de dados dedicado para o Runtime Store de cada implantação em um projeto Lakebase gerenciado pelo Databricks e o reutiliza quando você reimplanta. Você não pode usar seu próprio projeto Lakebase para o Runtime Store, e não o cria nem o vincula por conta própria. Os resultados e os eventos sobrevivem a reinicializações de worker, e qualquer instância do agente pode atender a solicitações de status e reconexão.agentbricks deployments deleteremove o Runtime Store junto com a implantação.
The Runtime Store holds the server's execution state. It's separate from the session and memory stores that your agent uses for conversation história and long-term memory.
Recuperação de falhas
Para recuperar execuções interrompidas por falhas ou reinicializações de um worker, registro um manipulador de recuperação com @app.recover. Quando um servidor implantado detecta que os batimentos cardíacos de uma execução pararam, ele começa uma tentativa de substituição em um worker disponível e chama o manipulador de recuperação com a entrada original.
@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 você não fizer o registro de um manipulador de recuperação, a recuperação automática será desativada e o servidor Logs um aviso quando começar.
A recuperação funciona da seguinte maneira:
- Quando a recuperação começa : cada execução ativa envia um heartbeat a cada poucos segundos. Se os heartbeats pararem, por exemplo, porque o worker trava, é reiniciado ou é substituído durante uma nova implantação, o servidor detecta a execução obsoleta em segundos e começa uma tentativa de substituição.
- Quando a recuperação não começa : se o seu manipulador gerar uma exceção, a invocação falhará e o servidor não a repetirá. A recuperação cobre workers interrompidos, mas não erros no código do seu agente.
- Número de tentativas : O servidor não limita o número de tentativas de recuperação. Cada tentativa de substituição aumenta
context.attemptem um. Para parar após um número de tentativas, marquecontext.attemptno seu manipulador de recuperação e gere um erro. - Recuperação manual : não é possível Trigger a recuperação manualmente. Reenviar uma solicitação com o mesmo ID de invocação retorna a invocação existente em vez de iniciar uma nova tentativa.
A recuperação pode realizar a execução do código do agente mais de uma vez para a mesma invocação. Uma tentativa interrompida pode já ter chamado sistemas externos antes de a tentativa de substituição começar, portanto, torne essas chamadas idempotentes.
Biblioteca AgentKit
DurableAgentServer faz parte da biblioteca AgentKit, databricks_agentkit, que o pacote databricks-agentbricks inclui. Os projetos que você cria com agentbricks init importam a partir dele. A biblioteca exporta os seguintes auxiliares:
Exportar | Descrição |
|---|---|
| O servidor do agente e o contexto que ele passa para os seus manipuladores de invocação e recuperação. |
| Um cliente para memória gerenciada e repositórios de sessões. Ele cria e obtém repositórios, e expõe as memórias e sessões dos repositórios como objetos |
| Configure o rastreamento do MLflow para o agente e comece um rastreamento em torno de uma unidade de trabalho. |
| Crie um SDK |
| Liste os serviços de modelo que o agente pode chamar por meio do Unity Gateway. |
The biblioteca also includes framework helpers in databricks_agentkit.langgraph and databricks_agentkit.openai, which the generated padrões use to connect each framework to the session store. For the memory and session APIs, see gerenciado agent memory and gerenciado agent sessions.
Request-user authorization
Por default, as ferramentas do seu agente são executadas com as permissões do Service Principal do aplicativo. To execução a tool with the permissions of the user who sent the request, declare user authorization in agent.toml:
-
For a gerenciada tool, set
auth = "user"on the tool entry. Theagentbricks tools addcomandos for MCP servers, sandboxes, and Genie Agents writeauth = "user"by default. Pass--auth appto use the app's identity instead. -
Para uma ferramenta que você escreve em código, declare o requisito e quaisquer escopos de API que o Agent Bricks não possa inferir:
Toml[auth.user]
required = true
additional_api_scopes = ["sql"]
Quando um agente exige autorização do usuário, o DurableAgentServer lê a credencial do usuário nos cabeçalhos de solicitação confiáveis do Databricks Apps e a mantém na memória apenas para a tentativa ativa. O Runtime Store não armazena a credencial. No seu manipulador, obtenha um cliente do workspace para o usuário de 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") retorna um cliente que usa o service principal do aplicativo. O resolvedor é fechado quando a execução é encerrada, portanto, chame-o dentro do manipulador em vez de armazenar o cliente. Quando você executa o agente localmente com agentbricks dev, client_for("user") usa suas credenciais locais.
Ao implantar, agentbricks deploy solicita os escopos de usuário do Databricks Apps necessários para suas ferramentas. Para adicionar escopos ausentes a um aplicativo existente, passe --allow-user-scope-update. Consulte Configurar autorização em um aplicativo do Databricks.
Request-user invocations use the same synchronous, transmissão, background, and reconnection APIs. Como o servidor não armazena a credencial do usuário, ele não pode recuperar uma invocação de usuário de solicitação interrompida. The replacement attempt fails with the MCP_USER_AUTH_RECOVERY_UNSUPPORTED error before your handlers run.
Add custom Endpoint
DurableAgentServer é um aplicativo FastAPI. Adicione rotas junto à API de invocação da mesma forma que as adiciona a qualquer aplicativo FastAPI:
@app.get("/status")
async def status() -> dict:
return {"ready": True}
Padrões de framework
agentbricks init gera dois diretórios:
agent/contém o código do seu framework: o modelo, os prompts e as ferramentas.runtime/contém o adaptador que conecta o framework aDurableAgentServer, e o ponto de entrada que registro os manipuladores de invocação e recuperação do adaptador.
O adaptador traduz cada invocação em uma chamada para o loop de agente do framework e traduz a saída do framework em eventos e um resultado. Ambos os padrões registram um manipulador de recuperação. O padrão do LangGraph é retomado a partir do seu último ponto de verificação no armazenamento de sessões, e o padrão do OpenAI Agents SDK reproduz a solicitação na mesma sessão. Para integrar um agente existente, adicione um adaptador e um ponto de entrada DurableAgentServer, e defina server = "agentbricks" na seção [agent] de agent.toml.
Limitações
- Não é possível alterar o servidor de agentes de uma implantação existente. Para alternar entre
DurableAgentServere seu próprio servidor, crie um novo projeto com a opçãoagentbricks init --serverdesejada e implante-o com um novo nome. - Changing the
serverfield inagent.tomldoesn't convert existing server code intoDurableAgentServer. - Request-user authorization requires
server = "agentbricks".