Servidor do agente

Um servidor de agentes é a biblioteca que transforma seu código de agente em um serviço. Ele envolve o loop do agente em um servidor HTTP, define a API que os clientes chamam para executar o agente, gerencia as conexões dos clientes e determina o que acontece quando uma execução é interrompida. O servidor agente roda no runtime do agente. Para entender como as camadas se encaixam, veja Deploy agents no Azure Databricks.

Servidores de agente no Azure Databricks

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

Servidor do agente Pacote API do cliente Execução durável Usado por
DurableAgentServer (recomendado) databricks_agentkit, no pacote databricks-agentbricks API de invocação em /api/invocations: execuções síncronas, de streaming e em segundo plano, com reconexão de fluxo Um Runtime Store que agentbricks deploy faz o provisionamento, além de recuperação após falha por meio de um handler de recuperação Projetos que você cria com a CLI Agent Bricks
LongRunningAgentServer (herdado) databricks_ai_bridge.long_running, no databricks-ai-bridge[agent-server] pacote OpenAI Responses API em /responses, com execuções em segundo plano e retomada de stream Estado de execução em um banco de dados Lakebase que você configurar. Após uma falha, uma nova tentativa prossegue com a execução diretamente do registro de eventos da tentativa interrompida. O agent-openai-advanced e os modelos de aplicativo agent-langgraph-advanced
MLflow AgentServer (herdado) mlflow.genai.agent_server, no pacote mlflow API de Respostas OpenAI em /responses: execuções síncronas e em streaming None Os modelos básicos de aplicativos, como agent-openai-agents-sdk

LongRunningAgentServer estende o MLflow AgentServer, e ambos servem agentes que implementam a interface MLflow ResponsesAgent . Para implantar e manter um agente que use um deles, consulte Execute agentes nos Aplicativos do Databricks usando o servidor de agentes herdados. Para consultar um agente em qualquer um desses servidores, consulte Consultar agentes implantados no Azure Databricks.

DurableAgentServer

DurableAgentServer é o servidor agente Agent Bricks. Ele envolve seu loop de agente em um servidor HTTP que atende à API de invocação, acompanha cada execução e recupera execuções que um crash ou reinício interrompe. Agentes que você cria com a CLI Agent Bricks usam DurableAgentServer por padrão.

O DurableAgentServer oferece:

  • Uma API para todos os modos de solicitação: síncrono, streaming e invocações em segundo plano, além de reconexão de transmissão, todas atendidas pelo mesmo identificador.
  • Invocações idempotentes: Um ID de invocação gerado pelo cliente garante que uma requisição retentada não inicie uma execução duplicada.
  • Sessões ordenadas: Invocações na mesma sessão são executadas uma de cada vez, em ordem.
  • Estado persistente de execução: quando implantado, status de execução, eventos e resultados sobrevivem aos reinícios dos trabalhados.
  • Recuperação após falha: o servidor detecta execuções interrompidas e inicia uma nova tentativa.
  • Autorização do usuário solicitante: Ferramentas podem agir com as permissões do usuário que enviou a solicitação.
  • Endpoints personalizados: DurableAgentServer é uma aplicação FastAPI, então você pode adicionar suas próprias rotas.

Requirements

DurableAgentServer possui os seguintes requisitos:

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

Registre seu agente

Quando você cria um projeto com agentbricks init, a CLI faz isso para você. O arquivo gerado runtime/main.py cria o servidor e registra os handlers de invocação e recuperação do template, então você só edita o código do agente em agent/. Siga as etapas desta seção para usar um agente existente ou para gravar seu próprio identificador.

Crie um DurableAgentServer e registre um handler de invocação assíncrono com @app.invoke. O handler recebe o input da requisiçã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 registrar um manipulador de invocação, e o servidor não começa sem um. O identificador atende a todos os modos de solicitação: o cliente escolhe se espera pelo resultado, recebe eventos de fluxo ou executa em segundo plano.

Para rodar o servidor localmente, inicie-o com agentbricks dev. Projetos que você cria com agentbricks init incluem um ponto de entrada que roda o servidor com o Uvicorn, e um arquivo app.yaml que inicia o mesmo ponto de entrada após a implantação.

Contexto de invocação

O segundo argumento do manipulador é um InvocationContext:

Attribute Description
invocation_id O ID que o cliente enviou para essa invocação.
session_id A sessão à qual a invocação pertence, ou None se o cliente não enviou uma.
attempt O número da tentativa. A primeira tentativa é 1.
is_recovery True quando o manipulador de recuperação está executando uma tentativa de substituição.
emit(event) Armazena um evento JSON, entrega para clientes de streaming e retorna a posição do evento no stream.
request_auth O resolvedor de credenciais do usuário da solicitação, quando o agente requer autorização do usuário da solicitação. Caso contrário, None.

API de invocação

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

  • POST /api/invocations inicia uma invocação. Por padrão, a solicitação aguarda o resultado. Defina stream para receber eventos como Server-Sent Eventos, ou background para retornar imediatamente com uma URL de status.
  • GET /api/invocations/<id> retorna o status de uma invocação e, após sua conclusão, sua saída.
  • GET /api/invocations/<id>/events?after=<event-id> transmite eventos armazenados, para que um cliente possa se reconectar após uma conexão caída.

Para obter campos de solicitação, exemplos e formatos de resposta, consulte Agentes de consulta implantados no Azure Databricks.

Idempotência

Os clientes enviam um UUID id a cada invocação. O servidor trata o ID como uma chave de idempotência enquanto mantém o registro de invocação: reenviar a mesma solicitação retorna a invocação existente em vez de rodar o agente novamente. Reutilizar um ID para uma requisição diferente retorna um erro 409.

Sessions

Os clientes podem enviar um session_id para agrupar invocações em uma única conversa. O servidor armazena o ID de sessão separadamente de input, o passa para seu handler 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 uma ID de sessão, uma invocação não estará associada a uma sessão.

Estado de execução

DurableAgentServer armazena a solicitação, o status, os sinais de pulsação, os eventos e o resultado de cada invocação em um Repositório de Runtime.

  • Desenvolvimento local: agentbricks dev usa um Repositório de Runtime. A API de invocação se comporta da mesma forma, mas o estado de execução é perdido quando o processo para e o servidor não reinicia o trabalho interrompido.
  • Agentes implantados: agentbricks deploy provisionam um banco de dados dedicado para o Repositório de Runtime de cada implantação em um projeto Lakebase gerenciado do Azure Databricks, e ele é reutilizado quando você reimplanta. Você não pode usar seu próprio projeto Lakebase para a Runtime Store, e você mesmo não o cria nem o vincula. Resultados e eventos sobrevivem a reinícios de trabalhadores, e qualquer instância do agente pode atender a solicitações de status e reconexão. agentbricks deployments delete remove o Repositório de Runtime com a implantação.

A Runtime Store mantém o estado de execução do servidor. É separado dos repositórios de sessão e memória que seu agente usa para histórico de conversas e memória de longo prazo.

Recuperação após falha

Para recuperar execuções que são interrompidas por uma falha ou reinicialização de um worker, registre um handler de recuperação com @app.recover. Quando um servidor implantado detecta que os heartbeats de uma execução pararam, ele inicia uma tentativa de substituição em um worker disponível e chama o handler 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 registrar um handler de recuperação, a recuperação automática fica desativada e o servidor registra um aviso quando ele começa.

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 as pulsações pararem, por exemplo, porque o trabalho trava, reinicia ou é substituído durante uma nova implantação, o servidor detecta a execução obsoleta em segundos e inicia uma nova tentativa em substituição.
  • Quando a recuperação não começa: Se seu handler levantar uma exceção, a invocação falha e o servidor não 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 em seu manipulador de recuperação e gere um erro.
  • Recuperação manual: Você não pode acionar 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 rodar seu código de agente mais de 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, então faça essas chamadas idempotentes.

Biblioteca AgentKit

DurableAgentServer faz parte da biblioteca do AgentKit, databricks_agentkit, incluída no pacote databricks-agentbricks. Os projetos que você cria com agentbricks init importados dele. A biblioteca exporta os seguintes auxiliares:

Export Description
DurableAgentServer, InvocationContext O servidor agente e o contexto que ele passa para seus manipuladores de invocação e recuperação.
AgentKitClient Um cliente para memória gerenciada e repositórios de sessão. Cria e obtém repositórios e expõe as memórias e sessões dos armazenamentos como objetos Memory, MemoryStore, MemorySearchResult, Session, SessionStore e SessionItem.
configure_tracing, start_trace Configure o rastreamento do MLflow para o agente e inicie um rastreamento em torno de uma unidade de trabalho.
workspace_client, workspace_headers Crie um SDK Databricks WorkspaceClient autenticado, ou obtenha cabeçalhos de autenticação para chamadas HTTP diretas, do ambiente do agente.
list_ai_gateway_model_services Liste os serviços modelo que o agente pode chamar pelo Unity Gateway.

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

Autorização de usuário da solicitação

Por padrão, as ferramentas do seu agente rodam com as permissões do principal de serviço do app. Para executar uma ferramenta com as permissões do usuário que enviou a solicitação, declare a autorização do usuário em agent.toml:

  • Para uma ferramenta gerenciada, defina auth = "user" no item da ferramenta. Os agentbricks tools add comandos para servidores MCP, sandboxes e Agentes Genie escrevem auth = "user" por padrão. Forneça --auth app para usar a identidade do aplicativo.

  • Para uma ferramenta que você escreve em código, declare o requisito necessário e quaisquer escopos de API que o Agent Bricks não possa inferir:

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

Quando um agente precisa de autorização do usuário, DurableAgentServer lê a credencial do usuário dos cabeçalhos de solicitação confiáveis do Databricks Apps e a mantém na memória apenas para a tentativa ativa. A Runtime Store não armazena a credencial. No seu handler, obtenha um cliente de workspace para o usuário 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") retorna um cliente que usa o principal de serviço do app. O resolver fecha quando a tentativa termina, então chame-o dentro do handler em vez de armazenar o cliente. Quando você executa o agente localmente com agentbricks dev, client_for("user") usa as credenciais locais.

Quando você implanta, agentbricks deploy solicita os escopos de usuário do Databricks Apps que suas ferramentas precisam. Para adicionar escopos faltantes a um aplicativo existente, passe --allow-user-scope-update. Consulte Configurar a autorização em um aplicativo do Databricks.

Invocações de request-user usam as mesmas APIs síncronas, de streaming, em segundo plano e de reconexão. Como o servidor não armazena a credencial do usuário, ele não pode recuperar uma invocação de solicitação do usuário interrompida. A tentativa de substituição falha com o erro MCP_USER_AUTH_RECOVERY_UNSUPPORTED antes de os identificadores serem executados.

Adicionar endpoints personalizados

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

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

Modelos de estrutura

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 a DurableAgentServer, e o ponto de entrada que registra os manipuladores de invocação e recuperação do adaptador.

O adaptador traduz cada invocação em uma chamada para o loop agente do framework e traduz a saída do framework em eventos e um resultado. Ambos os templates registram um handler de recuperação. O modelo LangGraph retoma a partir do último checkpoint no armazenamento de sessões, e o template do SDK dos Agentes OpenAI reproduz a solicitação na mesma sessão. Para incorporar um agente existente, adicione um adaptador e um entrypoint DurableAgentServer, e defina server = "agentbricks" na seção [agent] de agent.toml.

Limitações

  • Você não pode mudar o servidor agente de uma implantação existente. Para alternar entre DurableAgentServer e o seu próprio servidor, crie um novo projeto com a agentbricks init --server opção que você quiser e implante-o sob um novo nome.
  • Mudar o campo server em agent.toml não converte o código existente do servidor em DurableAgentServer.
  • A autorização do usuário da solicitação requer server = "agentbricks".

Recursos adicionais

  • Última atualização em