代理伺服器

代理伺服器是將你的代理程式碼轉化成服務的函式庫。 它將代理迴圈包裹在 HTTP 伺服器中，定義客戶端呼叫以執行代理的 API，管理客戶端連線，並決定執行遭中斷時會發生什麼。 代理伺服器在代理程式執行環境中執行。 想了解這些層如何相互配合，請參閱 在 Azure Databricks 上部署代理程式。

Azure Databricks 上的代理伺服器

Azure Databricks 提供三個代理伺服器。 對於新的代理程式，Databricks 建議使用 DurableAgentServer。

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

叫用 API

DurableAgentServer 提供位於 /api/invocations 的 invocation API：

• POST /api/invocations 開始叫用。 預設情況下，請求會等待結果。 設定 stream 為接收事件作為 Server-Sent 事件，或 background 立即返回並附上狀態網址。
• GET /api/invocations/<id> 回傳呼叫的狀態，完成後回傳其輸出。
• GET /api/invocations/<id>/events?after=<event-id> 串流傳送已儲存的事件，讓客戶端在連線中斷後能重新連線。

關於請求欄位、範例與回應格式，請參見 查詢部署於 Azure Databricks 的代理程式。

等冪性

客戶端每次呼叫都會傳送一個 UUID id 。 伺服器在保留呼叫記錄期間，會將 ID 視為冪等鍵：重複傳送相同請求時，會回傳現有的呼叫，而非重新執行代理。 重複使用一個 ID 來執行不同請求時，會傳回 409 錯誤。

會議

客戶端可以傳送 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 在部署時移除 Runtime Store。

Runtime Store 會儲存伺服器的執行狀態。 它和你的客服用來記錄對話歷史和長期記憶的 工作階段與記憶儲存區 是分開的。

當機復原

要恢復因工作者當機或重新啟動而中斷的執行，請使用 @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)

如果你沒有註冊復原處理程序，自動復原會被關閉，伺服器在啟動時會記錄警告。

復原過程如下：

• 恢復開始時：每次執行嘗試每隔幾秒就會發出一次心跳。 如果心跳停止，例如因為工作者當機、重啟或在重新部署時被替換，伺服器會在幾秒內偵測到過時的執行並啟動替代執行作業。
• 當復原未開始時：若處理器提出例外，呼叫失敗，伺服器不會重試該次呼叫。 復原涵蓋的是中斷的 worker，而非您的 Agent 程式碼中的錯誤。
• 嘗試次數：伺服器不會限制恢復次數。 每次更換嘗試都會使 context.attempt 增加 1。 若嘗試多次後停止，請在你的復原處理程式中檢查 context.attempt 並引發錯誤。
• 手動恢復：你無法手動觸發恢復。 重新傳送具有相同調用 ID 的請求時，會回傳現有的調用，而不是開始新的嘗試。

Recovery 可以針對同一呼叫多次執行你的代理程式程式碼。 中斷的嘗試可能在替換嘗試開始前已經呼叫了外部系統，因此應將這些呼叫設為冪等。

AgentKit 函式庫

DurableAgentServer 是 AgentKit 函式庫的一部分，databricks-agentbricks 套件包含 databricks_agentkit。 你使用 agentbricks init 從中匯入所建立的專案。 函式庫匯出以下輔助工具：

函式庫也包含位於 databricks_agentkit.langgraph 和 databricks_agentkit.openai 中的框架輔助工具，生成的範本會用來將每個框架連接到會話商店。 關於記憶體與工作階段 API，請參見 Managed agent 記憶體 與 Managed agent 工作階段。

請求使用者授權

預設情況下，你的代理工具會依應用程式服務主體的權限執行。 若要以發送請求使用者的權限執行工具，請在 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 應用程式一樣，將路徑與叫用 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 選項，並以新名稱部署。
• 在 agent.toml 中變更 server 欄位，並不會將現有的伺服器程式碼轉換成 DurableAgentServer。
• 要求使用者授權需要 server = "agentbricks"。

其他資源

• 代理執行時間
• 部署於 Azure Databricks 的查詢代理程式
• Agent Bricks CLI