Agent Server
エージェントサーバーとは、エージェントコードをサービスに変換するライブラリのことです。エージェントループをHTTPサーバー内にラップし、クライアントがエージェントを実行するために呼び出すAPIを定義し、クライアント接続を管理し、ランが中断されたときの動作を決定します。エージェントサーバーはエージェントランタイム上で実行されます。レイヤーがどのように組み合わされているかについては、Databricksでのエージェントのデプロイを参照してください。
Databricks上のエージェントサーバー
Databricks は3つのエージェントサーバーを提供しています。新しいエージェントの場合、Databricks では DurableAgentServer の使用を推奨しています。
エージェントサーバー | パッケージ | クライアント API | 耐久性のある実行 | 使用先 |
|---|---|---|---|---|
|
|
| Agent Bricks CLIで作成するプロジェクト | |
|
|
| 構成する Lakebase データベースにおけるランの状態。クラッシュ後、新しい試行によって、中断された試行のイベントLogsからランが続行されます。 |
|
MLflow |
|
| なし | ベースアプリTemplate（次など） |
LongRunningAgentServer は MLflow AgentServer を拡張し、両方とも MLflow ResponsesAgent インターフェイスを実装するエージェントを提供します。それらのいずれかを使用するエージェントをデプロイおよび保守するには、レガシーエージェントサーバーを使用して Databricks Apps でエージェントをランするを参照してください。これらのサーバーのいずれかでエージェントをクエリーするには、Databricks にデプロイされたエージェントをクエリーするを参照してください。
DurableAgentServer
DurableAgentServer は Agent Bricks エージェントサーバーです。エージェントのループをHTTPサーバーでラップし、呼び出しAPIの提供、各ランの追跡、クラッシュや再起動によって中断されたランのリカバリを行います。Agent Bricks CLI で作成したエージェントは、default で DurableAgentServer を使用します。
DurableAgentServer provides:
- すべてのリクエストモードに対応する単一のAPI : 同期、ストリーミング、バックグラウンドの呼び出しに加え、ストリームの再接続もすべて同じハンドラーによって処理されます。
- べき等な呼び出し : クライアントが生成した呼び出し ID により、リトライされたリクエストが重複するランを起動しないよう保証されます。
- 順序付きセッション ：同じセッション内の呼び出しは、順番に1つずつ実行されます。
- 永続ラン状態 : デプロイ時、ランのステータス、イベント、結果はワーカーの再起動後も保持されます。
- クラッシュリカバリ ：サーバーは中断されたランを検出し、代替試行を起動します。
- リクエストユーザー認証 ：ツールは、リクエストを送信したユーザーの権限で実行できます。
- カスタムEndpoint ：
DurableAgentServerはFastAPIアプリケーションであるため、独自のルートを追加できます。
要件
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}
登録できるinvocation handlerは1つであり、ハンドラーがないとサーバーは起動しません。ハンドラーはすべてのリクエストモードに対応しており、クライアントは結果を待つか、イベントをストリームするか、バックグラウンドで実行するかを選択できます。
サーバーをローカルで実行するには、agentbricks dev で起動します。agentbricks init で作成したプロジェクトには、Uvicorn でサーバーを実行するエントリーポイントと、デプロイ後に同じエントリーポイントを起動する app.yaml ファイルが含まれています。
呼び出しコンテキスト
ハンドラーの 2 番目の引数は InvocationContext です。
属性 | 説明 |
|---|---|
| この呼び出しに対してクライアントが送信した ID。 |
| 呼び出しが属するセッション、またはクライアントが送信しなかった場合は |
| 試行回数。最初の試行は |
|
|
| JSON イベントを保存し、ストリーミング クライアントに配信し、ストリーム内のイベントの位置を返します。 |
| エージェントが リクエストユーザー認証を必要とする場合のリクエストユーザー資格情報リゾルバー。それ以外の場合は、 |
Invocation API
DurableAgentServer は、/api/invocations で呼び出しAPIを提供します。
POST /api/invocations呼び出しを起動します。defaultでは、リクエストは結果を待ちます。イベントを Server-Sent Events として受信するにはstreamを設定し、ステータス URL とともに即座に応答するにはbackgroundを設定します。GET /api/invocations/<id>インボケーションのステータスを返し、完了後に出力を返します。GET /api/invocations/<id>/events?after=<event-id>保存されたイベントをストリームするため、クライアントは接続が切断された後に再接続できます。
リクエストフィールド、例、およびレスポンスのフォーマットについては、Databricks 上にデプロイされたエージェントのクエリーを参照してください。
べき等性
クライアントは呼び出しごとにUUID idを送信します。サーバーは、呼び出しレコードを保持している間、IDをべき等キーとして扱います。同じリクエストを再送信すると、エージェントを再実行するのではなく、既存の呼び出しが返されます。異なるリクエストでIDを再利用すると、409エラーが返されます。
セッション
クライアントは session_id を送信して、呼び出しを1つの会話にグループ化できます。サーバーはセッションIDを input とは別に保存し、context.session_id としてハンドラーに渡し、セッションIDを共有する呼び出しを1つずつ順番に実行します。サーバーは、呼び出しIDまたは入力からセッションを推論しません。セッションIDがない場合、呼び出しはセッションレスになります。
ランのステータス
DurableAgentServer 各呼び出しのリクエスト、ステータス、ハートビート、イベント、および結果をランタイムストア（Runtime Store）に保存します。
- ローカルでの開発 :
agentbricks devはインプロセス ランタイム ストアを使用します。呼び出し API の動作は同様ですが、プロセスが停止するとランのステータスが失われ、サーバーは中断された作業を再起動しません。 - デプロイ済みエージェント :
agentbricks deployは、Databricks が管理する Lakebase プロジェクト内の各デプロイの ランタイム Store 用に専用のデータベースをプロビジョニングし、再デプロイ時にそれを再利用します。ランタイム 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を確認し、エラーを発生させます。 - 手動リカバリ ：手動でリカバリをTriggerすることはできません。同じ呼び出し ID でリクエストを再送信すると、新しい試行を開始する代わりに既存の呼び出しが返されます。
リカバリは、同じ呼び出しに対してエージェントコードを2回以上ランする場合があります。中断された試行によって、置換の試行が起動される前に外部システムが既に呼び出されている可能性があるため、それらの呼び出しをべき等にしてください。
AgentKit ライブラリ
DurableAgentServer databricks-agentbricks パッケージに含まれる AgentKit ライブラリ (databricks_agentkit) の一部です。agentbricks init で作成したプロジェクトは、そこからインポートされます。ライブラリは、次のヘルパーをエクスポートします。
エクスポート | 説明 |
|---|---|
| エージェントサーバーと、それが呼び出しおよびリカバリハンドラーに渡すコンテキスト。 |
| マネージドメモリおよびセッションストア用のクライアント。ストアの作成および取得を行い、ストアのメモリとセッションを |
| エージェントの MLflow トレースを設定し、作業単位の周囲でトレースを起動します。 |
| エージェントの環境から、認証済みDatabricks SDK |
| エージェントが Unity Gateway を介して呼び出すことができるモデルサービスを一覧表示します。 |
ライブラリには、生成されたテンプレートが各フレームワークをセッションストアに接続するために使用する databricks_agentkit.langgraph および databricks_agentkit.openai のフレームワークヘルパーも含まれています。メモリおよびセッション APIs については、マネージドエージェントメモリと マネージドエージェントセッションを参照してください。
ユーザー承認をリクエストする
defaultでは、エージェントのツールはアプリの Service Principal の権限で実行されます。リクエストを送信したユーザーの権限でツールをランするには、agent.toml でユーザー認可を宣言します。
-
管理対象ツールの場合、ツールエントリで
auth = "user"を設定します。MCPサーバー、サンドボックス、およびGenieエージェント用のagentbricks tools addコマンドは、defaultでauth = "user"を書き込みます。代わりに--auth appを渡して、アプリのIDを使用します。 -
コードで記述するツールについては、要件とAgent Bricksで推論できないAPIスコープを宣言します。
Toml[auth.user]
required = true
additional_api_scopes = ["sql"]
エージェントにユーザー認可が必要な場合、DurableAgentServer は信頼できる Databricks Apps の要求ヘッダーからユーザーの資格情報を読み取り、アクティブな試行の間のみメモリに保持します。ランタイム 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") アプリのService Principalを使用するクライアントを返します。リゾルバーは試行の終了時にクローズするため、クライアントを保存する代わりにハンドラー内で呼び出してください。agentbricks dev と一緒にエージェントをローカルで実行する場合、client_for("user") はローカルの資格情報を使用します。
デプロイ時に、agentbricks deploy はツールに必要な Databricks Apps ユーザー スコープを要求します。既存のアプリに不足しているスコープを追加するには、--allow-user-scope-update を渡します。Databricks アプリで認可の設定をするを参照してください。
ユーザーリクエストによる呼び出しでは、同期、ストリーミング、バックグラウンド、および再接続の各APIが同じように使用されます。サーバーはユーザーの資格情報を保存しないため、中断されたユーザーリクエストによる呼び出しを復元することはできません。ハンドラーがランする前に、置換の試行が MCP_USER_AUTH_RECOVERY_UNSUPPORTED エラーで失敗します。
カスタムEndpointを追加する
DurableAgentServer は FastAPI アプリケーションです。任意の FastAPI アプリケーションに追加するのと同様の方法で、呼び出し API とともにルートを追加します。
@app.get("/status")
async def status() -> dict:
return {"ready": True}
フレームワーク Template
agentbricks init 2つのディレクトリを生成します。
agent/フレームワークコード（モデル、プロンプト、ツール）が含まれます。runtime/フレームワークをDurableAgentServerに接続するアダプターと、アダプターの invoke および recovery ハンドラーを登録するエントリーポイントが含まれます。
アダプターは各呼び出しをフレームワークのエージェントループへの呼び出しに変換し、フレームワークの出力をイベントと結果に変換します。両方のTemplateで、リカバリハンドラが登録されます。LangGraph Templateはセッションストアの最後のチェックポイントから再開し、OpenAI Agents SDK Templateは同じセッションでリクエストを再実行します。既存のエージェントを導入するには、アダプターと DurableAgentServer エントリポイントを追加し、agent.toml の [agent] セクションで server = "agentbricks" を設定します。
制限事項
- 既存のデプロイのエージェントサーバーは変更できません。
DurableAgentServerと独自のサーバーを切り替えるには、必要なagentbricks init --serverオプションを使用して新しいプロジェクトを作成し、新しい名前でデプロイします。 agent.toml内のserverフィールドを変更しても、既存のサーバー コードがDurableAgentServerに変換されるわけではありません。- リクエストユーザーの承認には
server = "agentbricks"が必要です。