Servidor de agente

Un servidor agente es la biblioteca que convierte tu código de agente en un servicio. Envuelve el bucle del agente en un servidor HTTP, define la API a la que llaman los clientes para ejecutar el agente, gestiona las conexiones del cliente y determina qué ocurre cuando una ejecución se interrumpe. El servidor del agente funciona en el entorno de ejecución del agente. Para saber cómo encajan las capas, consulta Desplegar agentes en Azure Databricks.

Servidores de agente en Azure Databricks

Azure Databricks proporciona tres servidores de agente. Para agentes nuevos, Databricks recomienda DurableAgentServer.

Servidor del agente Paquete API de cliente Ejecución duradera Usado por
DurableAgentServer (recomendado) databricks_agentkit, en el paquete databricks-agentbricks API de invocación en /api/invocations: ejecuciones síncronas, de streaming y en segundo plano, con reconexión de flujo Un Runtime Store que agentbricks deploy aprovisiona, además de la recuperación de bloqueos a través de un gestor de recuperación Proyectos que creas con la CLI de Agent Bricks
LongRunningAgentServer (heredado) databricks_ai_bridge.long_running, incluido en el paquete databricks-ai-bridge[agent-server] OpenAI Responses API en /responses, con ejecuciones en segundo plano y reanudación de flujos Estado de ejecución en una base de datos Lakebase que configures. Tras un fallo, un nuevo intento continúa la ejecución desde el registro de eventos del intento interrumpido. Las agent-openai-advanced y agent-langgraph-advanced
MLflow AgentServer (legado) mlflow.genai.agent_server, en el paquete mlflow API de OpenAI Responses en /responses ejecuciones síncronas y de streaming None Las plantillas base de aplicaciones, como agent-openai-agents-sdk

LongRunningAgentServer extiende el MLflow AgentServer, y ambos sirven a agentes que implementan la interfaz MLflow ResponsesAgent . Para desplegar y mantener un agente que utilice uno de ellos, consulta Ejecución de agentes en aplicaciones de Databricks mediante el servidor de agentes heredado. Para consultar un agente en cualquiera de estos servidores, consulte Consultar agentes desplegados en Azure Databricks.

DurableAgentServer

DurableAgentServer es el servidor agente Agent Bricks. Envuelve tu bucle de agentes en un servidor HTTP que sirve la API de invocación, rastrea cada ejecución y recupera las ejecuciones interrumpidas por un fallo o reinicio. Los agentes que creas con la CLI de Agent Bricks usan DurableAgentServer por defecto.

DurableAgentServer ofrece:

  • Una API para cada modo de solicitud: sincrónicas, en streaming y en segundo plano, además de reconexión de stream, todas atendidas por el mismo manejador.
  • Invocaciones idempotentes: Un ID de invocación generado por el cliente asegura que una solicitud reintentada no inicie una ejecución duplicada.
  • Sesiones ordenadas: Las invocaciones en la misma sesión se ejecutan una a una, en orden.
  • Estado de ejecución persistente: Cuando se despliega, el estado de la ejecución, los eventos y los resultados sobreviven a los reinicios de los trabajadores.
  • Recuperación tras fallos: El servidor detecta ejecuciones interrumpidas y comienza un intento de reemplazo.
  • Autorización del usuario solicitante: Las herramientas pueden actuar con los permisos del usuario que envió la solicitud.
  • Endpoints personalizados: DurableAgentServer es una aplicación FastAPI, así que puedes añadir tus propias rutas.

Requirements

DurableAgentServer Tiene los siguientes requisitos:

  • Python 3.10 y versiones posteriores.
  • El databricks-agentbricks paquete, que incluye la databricks_agentkit biblioteca. Los proyectos que creas con agentbricks init lo declaran como una dependencia.

Registra tu agente

Cuando creas un proyecto con agentbricks init, la CLI hace esto por ti. El generado runtime/main.py crea el servidor y registra los gestores de invocación y recuperación de la plantilla, así que solo editas el código del agente en agent/. Sigue los pasos de esta sección para incorporar un agente existente o para escribir tu propio handler.

Crea un DurableAgentServer y registra un manejador de invocación asíncrono con @app.invoke. El manejador recibe el input de la petición y un contexto de invocación, y devuelve un resultado serializable en JSON. Publica el progreso como eventos 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}

Puedes registrar un solo manejador de invocación y el servidor no arranca sin uno. El manejador admite todos los modos de solicitud: el cliente elige si esperar el resultado, transmitir eventos o ejecutarse en segundo plano.

Para ejecutar el servidor localmente, inícialo con agentbricks dev. Los proyectos que creas con agentbricks init incluyen un punto de entrada que ejecuta el servidor con Uvicorn y un archivo app.yaml que inicia el mismo punto de entrada después de desplegar.

Contexto de invocación

El segundo argumento del manejador es un InvocationContext:

Attribute Description
invocation_id El ID que el cliente envió para esta invocación.
session_id La sesión a la que pertenece la invocación, o None si el cliente no envió una.
attempt El número de intento. El primer intento es 1.
is_recovery True cuando el controlador de recuperación está ejecutando un intento de reemplazo.
emit(event) Almacena un evento JSON, lo entrega a clientes de streaming y devuelve la posición del evento en el stream.
request_auth El resolutor de credenciales del usuario solicitante, cuando el agente requiere autorización del usuario solicitante. En caso contrario, es None.

API de invocación

DurableAgentServer expone la API de invocación en /api/invocations:

  • POST /api/invocations inicia una invocación. Por defecto, la solicitud espera hasta obtener el resultado. Configura stream para recibir eventos como Server-Sent Eventos, o background para que devuelva inmediatamente una URL de estado.
  • GET /api/invocations/<id> devuelve el estado de una invocación y, una vez completada, su salida.
  • GET /api/invocations/<id>/events?after=<event-id> transmite eventos almacenados para que un cliente pueda reconectarse tras una conexión caída.

Para campos de solicitud, ejemplos y formatos de respuesta, véase Consultar agentes desplegados en Azure Databricks.

Idempotencia

Los clientes envían un UUID id con cada invocación. El servidor trata el ID como una clave de idempotencia mientras conserva el registro de invocación: reenviar la misma solicitud devuelve la invocación existente en lugar de ejecutar el agente de nuevo. Reutilizar un ID para una solicitud diferente devuelve un error 409.

Sessions

Los clientes pueden enviar una session_id para agrupar invocaciones en una sola conversación. El servidor almacena el ID de sesión por separado de input, lo pasa a tu manejador como context.session_id, y ejecuta invocaciones que comparten un ID de sesión una a una, en orden. El servidor no infiere una sesión a partir del ID de invocación ni de la entrada. Sin un ID de sesión, una invocación es sin sesión.

Estado de ejecución

DurableAgentServer almacena la solicitud, el estado, los latidos, los eventos y el resultado de cada invocación en un almacén de ejecución.

  • Desarrollo local: agentbricks dev utiliza un Runtime Store en proceso. La API de invocación se comporta igual, pero el estado de ejecución se pierde cuando el proceso se detiene y el servidor no reinicia el trabajo interrumpido.
  • Agentes desplegados: agentbricks deploy aprovisiona una base de datos dedicada para el Runtime Store de cada despliegue en un proyecto Lakebase gestionado por Azure Databricks, y la reutiliza cuando se vuelve a desplegar. No puedes usar tu propio proyecto Lakebase para la Runtime Store, ni lo creas ni lo enlazas tú mismo. Los resultados y eventos sobreviven a los reinicios de los trabajadores, y cualquier instancia del agente puede servir solicitudes de estado y reconexión. agentbricks deployments delete elimina Runtime Store junto con el despliegue.

El Runtime Store contiene el estado de ejecución del servidor. Está separado de los almacenes de sesiones y memoria que tu agente usa para el historial de conversaciones y la memoria a largo plazo.

Recuperación tras un bloqueo

Para recuperar las ejecuciones que se interrumpen por un bloqueo o reinicio de un worker, se registra un gestor de recuperación con @app.recover. Cuando un servidor desplegado detecta que los latidos de una ejecución se han detenido, inicia un intento de reemplazo sobre un trabajador disponible y llama al gestor de recuperación con la 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)

Si no registras un gestor de recuperación, la recuperación automática se desactiva y el servidor registra una advertencia al iniciar.

La recuperación funciona de la siguiente manera:

  • Cuando comienza la recuperación: Cada intento en ejecución envía un latido cada pocos segundos. Si los latidos se detienen, por ejemplo porque el trabajador falla, se reinicia o es reemplazado durante un redespliegue, el servidor detecta la ejecución obsoleta en cuestión de segundos y comienza un intento de reemplazo.
  • Cuando la recuperación no inicia: Si tu manejador lanza una excepción, la invocación falla y el servidor no la vuelve a intentar. La recuperación cubre a los workers interrumpidos, no a los errores en el código de tu agente.
  • Número de intentos: El servidor no limita el número de intentos de recuperación. Cada intento de reemplazo aumenta context.attempt en uno. Para parar tras varios intentos, comprueba context.attempt en tu controlador de recuperación y lanza un error.
  • Recuperación manual: No puedes activar la recuperación manualmente. Reenviar una solicitud con el mismo ID de invocación devuelve la invocación existente en lugar de iniciar un nuevo intento.

Recovery puede ejecutar tu código de agente más de una vez para la misma invocación. Un intento interrumpido podría haber llamado ya a sistemas externos antes de que comience el intento de reemplazo, así que haz que esas llamadas sean idempotentes.

Biblioteca AgentKit

DurableAgentServer forma parte de la biblioteca AgentKit, databricks_agentkitque incluye el databricks-agentbricks paquete. Los proyectos que creas con agentbricks init importan de ella. La biblioteca exporta las siguientes funciones auxiliares:

Export Description
DurableAgentServer, InvocationContext El servidor agente y el contexto que transmite a tus manejadores de invocación y recuperación.
AgentKitClient Un cliente para memoria gestionada y almacenes de sesión. Crea y obtiene almacenes, y expone las memorias y sesiones de los almacenes como objetos Memory, MemoryStore, MemorySearchResult, Session, SessionStore y SessionItem.
configure_tracing, start_trace Configura el trazado de MLflow para el agente y inicia una traza alrededor de una unidad de trabajo.
workspace_client, workspace_headers Crea un SDK WorkspaceClientde Databricks autenticado, o obtén encabezados de autenticación para llamadas HTTP directas, desde el entorno del agente.
list_ai_gateway_model_services Enumera los servicios de modelo que el agente puede invocar a través de Unity Gateway.

La biblioteca también incluye auxiliares de framework en databricks_agentkit.langgraph y databricks_agentkit.openai, que las plantillas generadas utilizan para conectar cada framework al almacén de sesiones. Para las APIs de memoria y sesión, véanse Memoria de agentes gestionados y Sesiones de agentes gestionados.

Autorización del usuario solicitante

Por defecto, las herramientas de tu agente se ejecutan con los permisos del principal de servicio de la app. Para ejecutar una herramienta con los permisos del usuario que envió la solicitud, declara la autorización de usuario en agent.toml:

  • Para una herramienta gestionada, pon auth = "user" en la entrada de herramienta. Los agentbricks tools add comandos para servidores MCP, sandboxes y agentes Genie escriben auth = "user" por defecto. Pasa --auth app para usar la identidad de la app en su lugar.

  • Para una herramienta que escribas en código, declara los requisitos y cualquier alcance de API que Agent Bricks no pueda inferir:

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

Cuando un agente requiere autorización de usuario, DurableAgentServer lee la credencial del usuario de los encabezados de solicitud de Databricks Apps de confianza y la mantiene en memoria solo para el intento activo. Runtime Store no almacena la credencial. En tu manejador, obtén un cliente de espacio de trabajo del usuario desde 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") Devuelve un cliente que utiliza la entidad de servicio de la aplicación. El resolver se cierra cuando termina el intento, así que llámalo dentro del manejador en lugar de almacenar el cliente. Cuando ejecutas el agente localmente con agentbricks dev, client_for("user") usa tus credenciales de tu entorno local.

Cuando despliegas, agentbricks deploy solicita los ámbitos de usuario de Databricks Apps que necesitan tus herramientas. Para añadir permisos faltantes a una aplicación existente, pasa --allow-user-scope-update. Consulte Configuración de la autorización en una aplicación de Databricks.

Las invocaciones de request-user utilizan las mismas APIs síncronas, de streaming, en segundo plano y de reconexión. Como el servidor no almacena la credencial del usuario, no puede recuperar una invocación request-user interrumpida. El intento de reemplazo falla con el error MCP_USER_AUTH_RECOVERY_UNSUPPORTED antes de que se ejecuten tus controladores.

Añadir endpoints personalizados

DurableAgentServer es una aplicación FastAPI. Añade rutas junto a la API de invocación de la misma manera que las añades a cualquier aplicación de FastAPI:

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

Plantillas de framework

agentbricks init genera dos directorios:

  • agent/ Contiene tu código de framework: el modelo, los prompts y las herramientas.
  • runtime/ contiene el adaptador que conecta el framework con DurableAgentServer, y el punto de entrada que registra los manejadores de invocación y recuperación del adaptador.

El adaptador traduce cada invocación en una llamada al bucle agente del framework, y traduce la salida del framework en eventos y un resultado. Ambas plantillas registran un gestor de recuperación. La plantilla de LangGraph se reanuda desde su último punto de control en el almacén de sesiones, y la plantilla del SDK de Agentes de OpenAI reproduce la solicitud en la misma sesión. Para incorporar un agente existente, añade un adaptador y un DurableAgentServer entrypoint, y establece server = "agentbricks" en la sección [agent] de agent.toml.

Limitaciones

  • No puedes cambiar el servidor del agente de un despliegue existente. Para cambiar entre DurableAgentServer y tu propio servidor, crea un nuevo proyecto con la opción agentbricks init --server que quieras y despléchalo con un nombre nuevo.
  • Cambiar el campo server en agent.toml no convierte el código existente del servidor en DurableAgentServer.
  • La autorización del usuario solicitante requiere server = "agentbricks".

Recursos adicionales

  • Se actualizó por última vez el