Agent Server

Um servidor de agentes é a biblioteca que transforma o seu código de agente num serviço. Envolve o ciclo do agente num servidor HTTP, define a API que os clientes invocam para executar o agente, gere as ligações dos clientes e determina o que acontece quando uma execução é interrompida. O servidor agente corre no runtime do agente. Para saber como as camadas se encaixam, veja Deploy agents no Azure Databricks.

Agent servers on Azure Databricks

Azure Databricks fornece três servidores agentes. Para agentes novos, o Databricks recomenda DurableAgentServer.

LongRunningAgentServer estende o MLflow AgentServer, e ambos servem agentes que implementam a interface MLflow ResponsesAgent . Para implementar e manter um agente que utilize um deles, veja Executar agentes em aplicações Databricks usando o servidor de agentes legado. Para consultar um agente em qualquer um destes servidores, consulte Consultar agentes implementados no Azure Databricks.

DurableAgentServer

DurableAgentServer é o servidor de agentes do Agent Bricks. Envolve o seu loop de agente num servidor HTTP que serve a API de invocação, acompanha cada execução e recupera as execuções que um crash ou reinício interrompe. Os agentes que crias com a CLI Agent Bricks usam DurableAgentServer por defeito.

DurableAgentServer Oferece:

• Uma API para cada modo de solicitação: Síncrono, de streaming e invocações em segundo plano, além da reconexão do fluxo, todas servidas pelo mesmo handler.
• Invocações idempotentes: Um ID de invocação gerado pelo cliente garante que um pedido retentado não inicia uma execução duplicada.
• Sessões ordenadas: As invocações na mesma sessão são executadas uma de cada vez, por ordem.
• Estado persistente de execução: Quando é implementado, o estado da execução, eventos e resultados sobrevivem aos reinícios dos trabalhadores.
• Recuperação em caso de falha: O servidor deteta execuções interrompidas e inicia uma nova tentativa.
• Autorização do utilizador que faz o pedido: As ferramentas podem agir com as permissões do utilizador que enviou o pedido.
• Endpoints personalizados: DurableAgentServer é uma aplicação FastAPI, por isso pode adicionar as suas próprias rotas.

Requirements

DurableAgentServer tem os seguintes requisitos:

• Python 3.10 e superiores.
• O databricks-agentbricks pacote, que inclui a databricks_agentkit biblioteca. Os projetos que crias com agentbricks init declaram-no como uma dependência.

Regista o teu agente

Quando crias um projeto com agentbricks init, a CLI faz isso por ti. O ficheiro gerado runtime/main.py cria o servidor e regista os handlers de invocação e recuperação do template, por isso só editas o código do agente em agent/. Siga os passos desta secção para integrar um agente existente ou para escrever o seu próprio handler.

Criar um DurableAgentServer e registar um handler de invocação assíncrono com @app.invoke. O handler recebe o input do pedido e um contexto de invocação, e retorna um resultado serializável em JSON. Publicar 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}

Pode registar um handler de invocações, e o servidor não inicia sem um. O manipulador suporta todos os modos de pedido: o cliente escolhe se espera pelo resultado, recebe eventos em fluxo ou executa em segundo plano.

Para correr o servidor localmente, inicie-o com agentbricks dev. Os projetos que crias com agentbricks init incluem um ponto de entrada que executa o servidor com o Uvicorn, e um ficheiro app.yaml que inicia o mesmo ponto de entrada depois de implementares.

Contexto de invocação

O segundo argumento do manipulador é um InvocationContext:

API de invocação

DurableAgentServer disponibiliza a API de invocação em /api/invocations:

• POST /api/invocations inicia uma invocação. Por predefinição, a solicitação aguarda o resultado. Definir stream para receber eventos como Server-Sent Events, ou background para devolver imediatamente com um URL de estado.
• GET /api/invocations/<id> retorna o estado de uma invocação e, após esta concluída, a sua saída.
• GET /api/invocations/<id>/events?after=<event-id> transmite eventos armazenados, para que um cliente possa reconectar-se após uma ligação interrompida.

Para campos da solicitação, exemplos e formatos de resposta, consulte Consultar agentes implementados no Azure Databricks.

Idempotência

Os clientes enviam um UUID id com cada invocação. O servidor trata o ID como uma chave de idempotência enquanto mantém o registo de invocação: reenviar o mesmo pedido devolve a invocação existente em vez de executar o agente novamente. Reutilizar um ID para um pedido diferente devolve um 409 erro.

Sessions

Os clientes podem enviar uma session_id para agrupar invocações numa só conversa. O servidor armazena o ID da sessão separadamente de input, passa-o ao seu handler como context.session_id, e executa invocações que partilham um ID de sessão uma de cada vez, por 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 não está associada a uma sessão.

Estado de execução

DurableAgentServer armazena o pedido, estado, batimentos cardíacos, eventos de cada invocação e resultado num Runtime Store.

• Desenvolvimento local: agentbricks dev utiliza uma Runtime Store no processo. A API de invocação comporta-se da mesma forma, mas o estado de execução perde-se quando o processo para e o servidor não reinicia o trabalho interrompido.
• Agentes implementados: agentbricks deploy prevê uma base de dados dedicada para o Runtime Store de cada implementação num projeto Lakebase gerido pelo Azure Databricks, e reutiliza-a quando se reimplementa. Não podes usar o teu próprio projeto Lakebase para a Runtime Store, e não o crias nem o associas tu próprio. Os resultados e eventos sobrevivem aos reinícios dos trabalhadores, e qualquer instância do agente pode fornecer pedidos de estado e de reconexão. agentbricks deployments delete remove o Runtime Store com a implantação.

O Runtime Store contém o estado de execução do servidor. É separado dos armazenamentos de sessão e memória que o teu agente usa para o histórico de conversas e memória de longo prazo.

Recuperação após falha

Para recuperar execuções que uma falha ou reinício de um worker interrompe, regista um handler de recuperação com @app.recover. Quando um servidor em produção deteta que os sinais de heartbeat de uma execução pararam, inicia uma tentativa de substituição num trabalhador disponível e chama o handler de recuperação com o input 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 não registares um processador de recuperação, a recuperação automática fica desligada e o servidor regista uma advertência no registo quando o servidor é iniciado.

A recuperação funciona da seguinte forma:

• Quando a recuperação começa: Cada tentativa em execução envia um heartbeat a cada poucos segundos. Se os heartbeats pararem, por exemplo porque o worker falha, reinicia ou é substituído durante uma nova implementação, o servidor deteta a execução desatualizada em segundos e inicia uma tentativa de substituição.
• Quando a recuperação não começa: Se o teu processador levantar uma exceção, a invocação falha e o servidor não a tenta novamente. A recuperação cobre workers interrompidos, 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.attempt em um. Para parar após um número de tentativas, verifique context.attempt no seu processador de recuperação e gere um erro.
• Recuperação manual: Não podes ativar a recuperação manualmente. Reenviar um pedido com o mesmo ID de invocação devolve a invocação existente em vez de iniciar uma nova tentativa.

A recuperação pode executar o código do seu agente mais do que uma vez para a mesma invocação. Uma tentativa interrompida pode já ter chamado sistemas externos antes do início da tentativa de substituição, por isso torna essas chamadas idempotentes.

Biblioteca AgentKit

DurableAgentServer faz parte da biblioteca AgentKit, databricks_agentkit, incluída no pacote databricks-agentbricks. Projetos que crias com agentbricks init importas dele. A biblioteca exporta as seguintes funções auxiliares:

A biblioteca inclui também auxiliares de framework em databricks_agentkit.langgraph e databricks_agentkit.openai, que os templates gerados usam para ligar cada framework ao armazenamento de sessões. Para as APIs de memória e de sessão, veja Memória de agente gerida e Sessões de agente gerido.

Pedir autorização do utilizador

Por defeito, as ferramentas do seu agente funcionam com as permissões do principal de serviço da aplicação. Para executar uma ferramenta com as permissões do utilizador que enviou o pedido, declare a autorização do utilizador em agent.toml:

• Para uma ferramenta gerida, defina auth = "user" no registo da ferramenta. Os agentbricks tools add comandos para servidores MCP, sandboxes e Agentes Genie escrevem auth = "user" por defeito. Forneça --auth app para usar a identidade da aplicação em vez disso.

• Para uma ferramenta que escreves em código, declara o requisito e quaisquer escopos de API que o Agent Bricks não consiga inferir:

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

Quando um agente necessita de autorização do utilizador, DurableAgentServer lê a credencial do utilizador a partir dos cabeçalhos de pedido confiáveis do Databricks Apps e mantém-na na memória apenas para a tentativa ativa. Runtime Store não armazena a credencial. No seu handler, obtenha um cliente de espaço de trabalho para o utilizador a partir 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") devolve um cliente que utiliza o service principal da aplicação. O resolvedor fecha quando a tentativa termina, por isso chama-o dentro do processador em vez de armazenar o cliente. Quando executas o agente localmente com agentbricks dev, client_for("user") usa as tuas credenciais locais.

Quando implementa, agentbricks deploy solicita os escopos de utilizador do Databricks Apps que as suas ferramentas precisam. Para adicionar âmbitos em falta a uma aplicação existente, passe --allow-user-scope-update. Consulte Configurar autorização em um aplicativo Databricks.

As invocações de request-user utilizam as mesmas APIs síncronas, de streaming, de segundo plano e de reconexão. Como o servidor não armazena a credencial do utilizador, não pode recuperar uma invocação pedido-utilizador interrompida. A tentativa de substituição falha com o erro MCP_USER_AUTH_RECOVERY_UNSUPPORTED antes de os teus handlers serem executados.

Adicionar endpoints personalizados

DurableAgentServer é uma aplicação FastAPI. Adicione rotas juntamente com a API de invocação da mesma forma que as adiciona a qualquer aplicação FastAPI:

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

Modelos de framework

agentbricks init gera dois diretórios:

• agent/ Contém o código do seu framework: o modelo, prompts e ferramentas.
• runtime/ contém o adaptador que liga a framework a DurableAgentServer, e o ponto de entrada que regista os handlers de invocação e recuperação do adaptador.

O adaptador traduz cada invocação numa chamada ao ciclo agente do framework e traduz a saída do framework em eventos e num resultado. Ambos os modelos registam um processador de recuperação. O template LangGraph retoma a partir do seu último checkpoint na loja de sessões, e o template do SDK dos Agentes OpenAI reexecuta o pedido na mesma sessão. Para trazer um agente existente, adicione um adaptador e um DurableAgentServer entrypoint, e defina server = "agentbricks" na [agent] secção de agent.toml.

Limitations

• Não podes alterar o servidor agente de uma implantação existente. Para alternar entre DurableAgentServer e o seu próprio servidor, crie um novo projeto com a opção agentbricks init --server que quiser e implemente-o com um novo nome.
• Mudar o server campo em agent.toml não converte o código existente do servidor em DurableAgentServer.
• A autorização do utilizador para pedidos requer server = "agentbricks".

Recursos adicionais

• Execução do Agente
• Consultar agentes implementados no Azure Databricks
• CLI do Agente Bricks