代理服务器

Agent 服务器是将你的代理代码转化为服务的库。 它将代理循环包裹在 HTTP 服务器中，定义客户端调用以运行代理的 API，管理客户端连接，并确定运行中断时的处理。 代理服务器运行在代理运行时。 想了解这些层如何相互衔接，请参见“在 Azure Databricks 上部署代理”。

Azure Databricks 上的代理服务器

Azure Databricks 提供三个代理服务器。 对于新智能体，Databricks 推荐 DurableAgentServer。

代理服务器	Package	客户端 API	持久执行	使用者
DurableAgentServer（推荐）	databricks_agentkit，在 databricks-agentbricks 包中	位于 /api/invocations 的调用 API：同步、流式和后台运行，并支持流重新连接	一个运行时存储器，负责配置 agentbricks deploy，并通过恢复处理程序进行崩溃恢复	你用 Agent Bricks CLI 创建的项目
LongRunningAgentServer（旧版）	databricks_ai_bridge.long_running，在 databricks-ai-bridge[agent-server] 包中	位于 /responses 的 OpenAI 响应 API，支持后台运行和流恢复	你所配置的 Lakebase 数据库的运行状态。 崩溃后，新的尝试会从中断尝试的事件日志继续运行。	agent-openai-advanced和agent-langgraph-advanced应用模板
MLflow AgentServer（旧版）	mlflow.genai.agent_server，在 mlflow 包中	OpenAI 响应 API 在 /responses：同步运行和流运行	None	基础应用模板，比如 agent-openai-agents-sdk

LongRunningAgentServer 扩展了 MLflow AgentServer，并且二者都为实现 MLflow ResponsesAgent 接口的代理提供服务。 要部署和维护使用其中一种代理服务器的代理，请参见 “使用遗留代理服务器在 Databricks Apps 上运行代理”。 要查询这些服务器上的代理，请参见“在 Azure Databricks 上部署的代理查询”。

DurableAgentServer

DurableAgentServer 是Agent Bricks的代理服务器。 它将你的代理循环包裹在一个 HTTP 服务器中，该服务器提供调用 API，跟踪每次运行，并恢复因崩溃或重启而中断的运行。 你用 Agent Bricks CLI 创建的代理默认使用 DurableAgentServer 。

DurableAgentServer 可以提供：

• 一个 API 适用于所有请求模式：同步、流式和后台调用，加上流重新连接，所有这些都由同一个处理器处理。
• 幂等调用请求：客户端生成的调用 ID 可确保重试请求不会启动重复运行。
• 有序会话：同一会话中的调用按顺序逐个运行。
• 持续运行状态：部署后，运行状态、事件和结果在工作线程重启后仍会保留。
• 崩溃恢复：服务器检测到被中断的运行并开始重新尝试。
• 请求-用户授权：工具可以根据发送请求的用户权限进行操作。
• 自定义端点： DurableAgentServer 是一个FastAPI应用，所以你可以添加自己的路由。

Requirements

DurableAgentServer 具有以下要求：

• Python 3.10及以上版本。
• 这个 databricks-agentbricks 套件，包括 databricks_agentkit 库。 你使用 agentbricks init 创建的项目会将其声明为依赖项。

注册你的代理

当你使用 agentbricks init 创建项目时，CLI 会帮你完成这件事。 生成的runtime/main.py会创建服务器并注册模板的调用和恢复处理程序，所以你只需在agent/中编辑代理代码。 请按照本节步骤引入现有代理，或自行编写处理程序。

创建一个DurableAgentServer，并使用@app.invoke注册一个异步调用处理程序。 处理器接收请求的 input 和调用上下文，并返回可序列化为 JSON 的结果。 使用 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}

你可以注册一个调用处理器，服务器没有调用处理器就不会启动。 处理器支持所有请求模式：客户端可选择等待结果、流式接收事件或在后台运行。

要在本地运行服务器，请使用 agentbricks dev 启动它。 你使用 agentbricks init 创建的项目包括一个用 Uvicorn 运行服务器的入口点，以及一个在部署后用于启动同一入口点的 app.yaml 文件。

调用上下文

处理程序的第二个参数是 InvocationContext：

属性	Description
invocation_id	客户端为此次调用发送的 ID。
session_id	调用所属的会话；如果客户端未发送会话，则为 None。
attempt	尝试编号。 第一次尝试是1。
is_recovery	True 当恢复处理程序执行替换尝试时。
emit(event)	存储一个 JSON 事件，将其传递给流式客户端，并返回事件在流中的位置。
request_auth	当代理需要请求-用户授权时使用的请求-用户凭证解析器 否则为 None。

调用接口 API

DurableAgentServer 在 /api/invocations 提供调用 API：

• POST /api/invocations 开始一次调用。 默认情况下，请求会等待结果。 将 stream 设置为以 Server-Sent Events 接收事件，或将 background 设置为立即返回并附带状态 URL。
• GET /api/invocations/<id> 返回调用的状态，完成后返回其输出。
• GET /api/invocations/<id>/events?after=<event-id> 流式传输已存储的事件，因此客户端在断开连接后可以重新连接。

关于请求字段、示例和响应格式，请参见 查询在 Azure Databricks 上部署的代理。

幂等性

客户端每次调用都会发送一个UUID id 。 服务器将ID视为幂等性密钥，同时保留调用记录：重复发送同一请求返回已有调用，而无需重新运行代理。 为不同请求重复使用ID会返回 409 错误。

Sessions

客户端可以发送一个 session_id，将调用归入同一个对话中。 服务器将会话 ID 与 input 分开存储，并将其作为 context.session_id 传递给你的处理器，并按顺序一次运行共享同一会话 ID 的调用。 服务器不会根据调用ID或输入推断会话。 没有会话ID，调用就是无会话的。

运行状态

DurableAgentServer 将每个调用的请求、状态、心跳、事件和结果存储在运行时存储中。

• 本地开发：agentbricks dev使用进程内 Runtime Store。 调用 API 的行为相同，但当进程停止时，运行状态会丢失，并且服务器不会重启被中断的工作。
• 部署代理：agentbricks deploy系统会在 Azure Databricks 管理的 Lakebase 项目中为每个部署的运行时存储预配专用数据库，并在重新部署时重用该数据库。 你不能使用自己的 Lakebase 项目作为 Runtime Store，且不是由你自行创建或绑定它。 结果和事件在工人重启后依然存在，任何代理实例都可以处理状态和重新连接请求。 agentbricks deployments delete 在部署时移除运行时存储。

运行时存储存储服务器的执行状态。 它与你的代理用于对话历史和长期记忆的 会话存储和记忆存储 是分开的。

崩溃恢复

要恢复因工作器崩溃或重启而中断的运行，使用 @app.recover 注册恢复处理程序。 当部署的服务器检测到某次运行实例的心跳停止时，会在可用工作器上启动替换尝试，并以原始输入呼叫恢复处理器。

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

如果你没有注册恢复处理程序，自动恢复会被关闭，服务器启动时会记录警告。

恢复过程如下：

• 恢复开始时：每次运行尝试每隔几秒发送一次心跳。 如果心跳停止，例如因为工作者崩溃、重启或在重新部署时被替换，服务器会在几秒钟内检测到失效的运行实例并开始替换尝试。
• 恢复未启动时：如果处理程序引发异常，调用将失败，服务器不会重试该调用。 恢复涵盖的是被中断的工作线程，而不是代理代码中的错误。
• 尝试次数：服务器不限制恢复尝试次数。 每次更换尝试都会使 context.attempt 增加 1。 要在多次尝试后停止，请在恢复程序中查看 context.attempt 并引发错误。
• 手动恢复：你不能手动触发恢复。 重新发送使用相同调用 ID 的请求时，会返回现有调用，而不是发起新的尝试。

对于同一次调用，恢复可能会多次运行代理代码。 中断的尝试可能在替换尝试开始前已经调用了外部系统，因此应使这些调用成为幂等。

AgentKit 库

DurableAgentServer 是 AgentKit 库 databricks_agentkit 的一部分，而 databricks-agentbricks 软件包包含它。 你创建的项目会从中通过 agentbricks init 导入。 该库导出以下辅助工具：

Export	Description
DurableAgentServer、InvocationContext	代理服务器及其传递给你的调用和恢复处理程序的上下文。
AgentKitClient	一个用于内存管理和会话存储的客户端。 它创建并获取存储，并将这些存储的内存和会话公开为 Memory、MemoryStore、SessionStore、MemorySearchResult、Session 和 SessionItem 对象。
configure_tracing、start_trace	为代理设置MLflow追踪，并围绕一个工作单元开始追踪。
workspace_client、workspace_headers	创建一个经过身份验证的 Databricks SDK 客户端 WorkspaceClient，或者从代理的环境中获取用于直接 HTTP 调用的身份验证头。
list_ai_gateway_model_services	列出代理可以通过 Unity Gateway 调用的模型服务。

库还包含位于databricks_agentkit.langgraph和databricks_agentkit.openai中的框架辅助工具，生成的模板使用这些辅助工具将各个框架连接到会话存储。 关于内存和会话API，请参见托管代理内存和托管代理会话。

请求用户授权

默认情况下，你的代理工具运行时会获得应用服务主体的权限。 要运行一个具有发送请求的用户权限的工具，请在 agent.toml 中声明用户授权：

• 对于托管工具，在工具条目上设置 auth = "user"。 agentbricks tools add MCP服务器、沙箱和Genie代理的命令默认将auth = "user"写入。 改为传入 --auth app 以使用应用的标识。

• 对于你用代码编写的工具，声明需求和Agent Bricks无法推断的API作用域：

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

当代理需要用户授权时， DurableAgentServer 会从受信任的Databricks Apps请求头读取用户凭证，并仅在当前尝试时将其保存在内存中。 Runtime Store 不会存储凭证。 在你的处理器中，为该用户从 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") 返回使用应用服务主体的客户端。 resolver 会在此次操作结束时关闭，因此应在处理器内部调用它，而不是存储客户端。 当你在本地运行代理 agentbricks dev时， client_for("user") 使用你的本地凭证。

部署时，agentbricks deploy 会请求你的工具所需的 Databricks Apps 用户范围。 要为现有应用添加缺失的作用域，请传递 --allow-user-scope-update。 请参阅 Databricks 应用中配置授权。

Request-user 调用使用相同的同步、流式、后台和重连 API。 由于服务器不存储用户凭证，因此无法恢复中断的请求用户调用。 替换尝试在您的处理程序运行之前因 MCP_USER_AUTH_RECOVERY_UNSUPPORTED 错误而失败。

添加自定义端点

DurableAgentServer 是一个FastAPI应用程序。 像添加到任何 FastAPI 应用一样，同时为 invocation API 添加路由：

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

框架模板

agentbricks init 生成两个目录：

• agent/ 包含你的框架代码：模型、提示和工具。
• runtime/ 包含连接框架到 DurableAgentServer的适配器，以及注册适配器调用和恢复处理程序的入口点。

适配器将每次调用转换成对框架代理循环的调用，并将框架输出转换为事件和结果。 两个模板都注册了恢复处理程序。 LangGraph 模板会从会话存储的最后检查点恢复，OpenAI Agents SDK 模板会在同一会话中重放请求。 要引入现有代理，请添加适配器和DurableAgentServer入口点，并在[agent]的server = "agentbricks"部分中设置agent.toml。

Limitations

• 你不能更改现有部署的代理服务器。 要在 DurableAgentServer 和你自己的服务器之间切换，使用你想要的 agentbricks init --server 选项创建一个新项目，并以新名称部署。
• 在server中更改agent.toml字段并不会将现有服务器代码转换为DurableAgentServer。
• 请求用户授权需要 server = "agentbricks"。

其他资源

• 代理运行时
• 部署在 Azure Databricks 上的查询代理
• Agent Bricks CLI