エージェントサーバーとは、エージェントコードをサービス化するライブラリのことです。 エージェントループをHTTPサーバーでラップし、クライアントがエージェントを実行するために呼び出すAPIを定義し、クライアントの接続を管理し、実行が中断されたときに何が起こるかを決定します。 エージェントサーバーはエージェントランタイム上で動作します。 レイヤーがどのように組み合わさるかについては、「Azure Databricks にエージェントをデプロイする」をご覧ください。
Azure Databricks 上のエージェント サーバー
Azure Databricksは3つのエージェントサーバーを提供しています。 新規エージェントには、Databricksが DurableAgentServerを推奨しています。
| エージェントサーバー | Package | クライアント API | 耐久性のある実行 | によって使用される |
|---|---|---|---|---|
DurableAgentServer (推奨) |
databricks_agentkit、databricks-agentbricks パッケージ内 |
/api/invocationsでの呼び出しAPI: 同期、ストリーミング、バックグラウンド実行、ストリーム再接続対応 |
agentbricks deploy プロビジョニングを行うランタイム ストアと、リカバリーハンドラーによるクラッシュ リカバリー |
Agent Bricks CLIで作成するプロジェクト |
LongRunningAgentServer (レガシー) |
databricks-ai-bridge[agent-server] パッケージ内の databricks_ai_bridge.long_running |
/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 はエージェントブリックスのエージェントサーバーです。 エージェントループをHTTPサーバーでラップし、呼び出しAPIを提供し、各実行を追跡し、クラッシュや再起動によって中断された実行を復旧します。
Agent Bricks CLIで作成したエージェントはデフォルトでDurableAgentServerを使用します。
DurableAgentServer は以下を提供します。
- すべてのリクエストモードに対して1つのAPI：同期、ストリーミング、バックグラウンド呼び出し、ストリーム再接続など、すべて同じハンドラで提供します。
- 冪等呼出:クライアント生成の呼出IDは、再試行リクエストが重複実行を開始しないようにします。
- 順序付きセッション:同じセッション内の呼び出しは順番に1つずつ実行されます。
- 永続的な実行状態: 展開時、実行状況、イベント、結果はワーカーの再起動後も保持されます。
- クラッシュリカバリー: サーバーは中断された実行を検知し、再試行を開始します。
- リクエスト-ユーザー認可: ツールはリクエストを送信したユーザーの権限に基づいて行動できます。
-
カスタムエンドポイント:
DurableAgentServerはFastAPIアプリケーションなので、自分でルートを追加できます。
Requirements
DurableAgentServer には次の要件があります:
- Python 3.10以降です。
-
databricks-agentbricksライブラリを含むdatabricks_agentkitパッケージです。agentbricks initで作成したプロジェクトは、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}
1つの呼び出しハンドラーを登録できます。サーバーはそれなしでは起動しません。 ハンドラーはすべてのリクエストモードに対応します。クライアントは結果を待つか、イベントをストリーミングするか、バックグラウンドで実行するかを選びます。
ローカルでサーバーを動かすには、 agentbricks devで起動してください。
agentbricks initで作成するプロジェクトには、Uvicornでサーバーを動かすエントリポイントと、デプロイ後に同じエントリポイントを起動するapp.yamlファイルが含まれます。
呼び出しコンテキスト
ハンドラーの第2引数は InvocationContextです。
| Attribute | 説明 |
|---|---|
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 を送信して呼び出しを1つの会話にまとめることができます。 サーバーはセッションIDを inputとは別に保存し、context.session_idとしてハンドラーに渡し、セッションIDを共有する呼び出しを順番に1つずつ実行します。 サーバーは呼び出しIDや入力からセッションを推定しません。 セッションIDがなければ、呼び出しはセッションレスです。
実行状態
DurableAgentServer 各呼び出しのリクエスト、ステータス、心拍、イベント、結果をランタイムストアに格納します。
-
ローカル開発:
agentbricks devインプロセス Runtime Store を使用しています。 呼び出しAPIも同じように動作しますが、プロセスが停止すると実行状態が失われ、中断された処理はサーバーによって再開されません。 -
デプロイ済みエージェント では、
agentbricks deployAzure Databricks管理のLakebaseプロジェクト内で、各デプロイメントのRuntime Store専用データベースをプロビジョニングし、再デプロイ時に再利用します。 Runtime Storeで自分のLakebaseプロジェクトを使うことはできませんし、自分で作成やバインドもできません。 結果やイベントはワーカーの再起動後も残り、エージェントの任意のインスタンスはステータスおよび再接続要求に応答できます。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 は、databricks-agentbricks パッケージに含まれる databricks_agentkit AgentKitライブラリの一部です。
agentbricks initからインポートして作成したプロジェクト。 ライブラリは以下のヘルパーをエクスポートしています:
| Export | 説明 |
|---|---|
DurableAgentServer、InvocationContext |
エージェントサーバーと、それがインヴォークやリカバリーハンドラーに渡すコンテキストです。 |
AgentKitClient |
マネージドメモリとセッションストア用のクライアントです。 ストアを作成・取得し、ストアのメモリやセッションを Memory、MemoryStore、MemorySearchResult、Session、SessionStore、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"を設定してください。 MCPサーバー、サンドボックス、Genie Agentsのagentbricks tools addコマンドはデフォルトでauth = "user"を書き込みます。 代わりにアプリの識別情報を使用するには、--auth appを渡します。コードで書くツールについては、要件とAgent Bricksが推論できないAPIスコープを宣言してください:
[auth.user] required = true additional_api_scopes = ["sql"]
エージェントがユーザーの認可を必要とする場合、DurableAgentServer は Databricks Apps の信頼されたリクエスト ヘッダーからユーザーの認証情報を読み取り、現在の試行中に限りメモリに保持します。 ランタイムストアには認証情報は保存されません。 ハンドラー内で、ユーザー用のワークスペースクライアントを 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 アプリでの承認の構成を参照してください。
リクエストユーザー呼び出しは、同じ同期、ストリーミング、バックグラウンド、再接続のAPIを使用します。 サーバーはユーザーの認証情報を保存しないため、中断されたrequest-user の呼び出しを復元できません。 ハンドラーが実行される前に MCP_USER_AUTH_RECOVERY_UNSUPPORTED エラーで代替試行が失敗します。
カスタムエンドポイントの追加
DurableAgentServer は FastAPIアプリケーションです。 任意の FastAPI アプリに追加するのと同じ方法で、呼び出し API と並行してルートを追加してください:
@app.get("/status")
async def status() -> dict:
return {"ready": True}
フレームワークテンプレート
agentbricks init 2つのディレクトリを生成する:
-
agent/には、フレームワークコード、モデル、プロンプト、ツールが含まれています。 -
runtime/には、フレームワークをDurableAgentServerに接続するアダプターと、アダプターの呼び出しおよび回復ハンドラを登録するエントリポイントが含まれます。
アダプターは各呼び出しをフレームワークのエージェントループへの呼び出しに変換し、フレームワークの出力をイベントと結果に変換します。 両方のテンプレートはリカバリーハンドラーを登録します。 LangGraphテンプレートはセッションストアの最後のチェックポイントから再開され、OpenAI Agents SDKテンプレートは同じセッション内でリクエストを再実行します。 既存のエージェントを導入するには、アダプターとDurableAgentServerエントリーポイントを追加し、server = "agentbricks"の[agent]セクションでagent.tomlを設定してください。
制限事項
- 既存のデプロイメントのエージェントサーバーを変更することはできません。
DurableAgentServerと自分のサーバーの間で切り替えるには、希望するagentbricks init --serverオプションで新しいプロジェクトを作成し、新しい名前でデプロイしてください。 -
serverのagent.tomlフィールドを変更しても、既存のサーバーコードがDurableAgentServerに変換されるわけではありません。 - リクエストユーザー認可には
server = "agentbricks"が必要です。