エージェントサーバー

エージェントサーバーとは、エージェントコードをサービス化するライブラリのことです。 エージェントループをHTTPサーバーでラップし、クライアントがエージェントを実行するために呼び出すAPIを定義し、クライアントの接続を管理し、実行が中断されたときに何が起こるかを決定します。 エージェントサーバーはエージェントランタイム上で動作します。 レイヤーがどのように組み合わさるかについては、「Azure Databricks にエージェントをデプロイする」をご覧ください。

新しいエージェントの場合、Databricks は AgentKit が提供するエージェント サーバーを推奨します。それについては、このページで説明しています。 それをレガシー エージェント サーバーと比較するには、「Azure Databricks のエージェント サーバーの比較」を参照してください。

Requirements

• Python 3.10以降です。
• databricks-agentbricksパッケージにはAgentKitが含まれています。 agentbricks init を使用して作成するプロジェクトでは、これを依存関係として指定します。

AgentKit

AgentKitは、Agent Runtime上で動作するエージェントを構築するためのAgent Bricksライブラリです。 エージェントを提供するエージェント サーバーに加え、エージェント メモリ、エージェント セッション、トレーシング、認証に対応するクライアントとヘルパーを提供します。 Agent Bricks CLIで作成したプロジェクトは、デフォルトでAgentKitを使用します。

エージェントサーバー

AgentKitは DurableAgentServer クラスを通じてエージェントサーバーを提供します。 エージェント サーバーは、エージェント ループを HTTP サーバーでラップし、そのサーバーが呼び出し用 API を提供するとともに、各実行を追跡し、クラッシュや再起動によって中断された実行を回復します。 次の機能を提供します。

• すべてのリクエストモードに対して1つのAPI：同期、ストリーミング、バックグラウンド呼び出し、ストリーム再接続など、すべて同じハンドラで提供します。
• 冪等呼出:クライアント生成の呼出IDは、再試行リクエストが重複実行を開始しないようにします。
• 順序付きセッション:同じセッション内の呼び出しは順番に1つずつ実行されます。
• 永続的な実行状態: デプロイされると、実行状態、イベント、結果は、ワーカーの再起動後も保持されます。
• クラッシュリカバリー: サーバーは中断された実行を検知し、再試行を開始します。
• リクエスト-ユーザー認可: ツールはリクエストを送信したユーザーの権限に基づいて行動できます。
• カスタムエンドポイント:エージェントサーバーはFastAPIアプリケーションなので、自分でルートを追加できます。

AgentKitに含まれるもの

AgentKit は databricks_agentkit Python ライブラリです。 以下のものをエクスポートします:

ライブラリには databricks_agentkit.langgraph および databricks_agentkit.openaiのフレームワークヘルパーも含まれており、生成されたテンプレートはそれぞれのフレームワークをセッションストアに接続するためにこれを利用しています。 メモリおよびセッションAPIについては、 エージェントメモリ および エージェントセッションを参照してください。

エージェントを登録

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 です:

呼び出し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 deploy は、Azure 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でリクエストを再送信すると、新しい試行を開始する代わりに既存の呼び出しが返されます。

リカバリーにより、同じ呼び出しに対してエージェント コードが複数回実行されることがあります。 中断された処理では、置換処理が開始される前にすでに外部システムへの呼び出しが行われている可能性があるため、それらの呼び出しをべき等性を持たせるようにしてください。

ユーザー認証の要求

デフォルトでは、エージェントのツールはアプリのサービスプリンシパルの権限で動作します。 リクエストを送信したユーザーの権限を持つツールを実行するには、 agent.tomlでユーザー認可を宣言してください:

• 管理ツールの場合は、ツールエントリで auth = "user" を設定してください。 MCP サーバー、サンドボックス、Genie エージェント向けの 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") アプリのサービスプリンシパルを使うクライアントを返します。 リソルバーは処理が終了すると閉じられるため、クライアントを保存するのではなく、ハンドラー内で呼び出してください。 エージェントをagentbricks devでローカルで実行すると、client_for("user")はローカルの認証情報を使用します。

デプロイを行うと、agentbricks deploy は、ご利用のツールに必要な Databricks Apps のユーザースコープを要求します。 既存のアプリに欠落しているスコープを追加するには、--allow-user-scope-update を指定してください。 Databricks アプリでの承認の構成を参照してください。

要求ユーザーの呼び出しでは、同期、ストリーミング、バックグラウンド再接続の各 API が同様に使用されます。 サーバーはユーザーの資格情報を保存していないため、中断された要求とユーザーの呼び出しを復元することはできません。 ハンドラーが実行される前に 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を設定してください。

Azure Databricks 上のエージェントサーバーを比較してください

Azure Databricksは3つのエージェントサーバーを提供しています:AgentKitエージェントサーバーと、以前のアプリテンプレートで使われていた2つのレガシーエージェントサーバーです。

LongRunningAgentServerはMLflow AgentServerを拡張し、両者ともMLflow ResponsesAgent インターフェースを実装するエージェントに対応します。 そのいずれかを使用するエージェントを展開・保守するには、「レガシーエージェントサーバーを使って Databricks Apps 上でエージェントを実行する」をご覧ください。 これらのサーバーのいずれかにデプロイされているエージェントをクエリするには、Azure Databricks にデプロイされたエージェントのクエリを参照してください。

制限事項

• 既存のデプロイメントのエージェントサーバーを変更することはできません。 DurableAgentServer とご利用のサーバーを切り替えるには、希望の agentbricks init --server オプションを指定して新しいプロジェクトを作成し、新しい名前でデプロイしてください。
• agent.toml の server フィールドを変更しても、既存のサーバーコードが DurableAgentServer に変換されることはありません。
• 要求ユーザーの承認には server = "agentbricks" が必要です。

その他のリソース

• エージェント ランタイム
• Azure Databricks にデプロイされたクエリ エージェント
• Agent Bricks CLI