Azure Databricks にデプロイされたクエリエージェント

エージェントのクエリ方法は、そのエージェントを提供するサーバーによって異なります。以下の表でエージェントを見つけ、該当するセクションに従ってください。エージェントサーバーについて知りたい方は、エージェントサーバーをご覧ください。

あなたのエージェント	Premium でホスト	クエリを実行する方法
`DurableAgentServer` を使用する	Databricks Apps 上の Agent Runtime	`/api/invocations` の Invocation API
MLflow`AgentServer`または`LongRunningAgentServer`を使用 (レガシー)	Databricks Apps	Databricks OpenAIクライアントまたは`/responses`のOpenAI応答API
モデルサービングにデプロイ済み（レガシー）	モデルサービングエンドポイント	Databricks OpenAIクライアント、REST API、 `ai_query`、またはAI Playground

Databricks Apps 上でホストされているエージェントは、 Databricks OAuth トークンが必要です。個人アクセストークンはDatabricksアプリでは機能しません。スクリプト、サービスプリンシパル、他のアプリ、またはノートブックからOAuthトークンを生成するには、トークン認証を使ったAPI Databricksアプリへの接続を参照してください。

`DurableAgentServer`を使用するエージェントをクエリする

DurableAgentServer を使用するエージェントは、呼び出し API を提供します。 Agent Bricks CLI を使用して作成したエージェントは DurableAgentServer を使用し、agentbricks deploy はそれらを Agent Runtime に agent-bricks-<name> という名前のアプリとしてデプロイします。 APIへの各リクエストは、エージェントの1回の実行を開始します。これはインボケーションと呼ばれます。

エンドポイント	説明
`POST /api/invocations`	呼び出しが始まります。デフォルトでは、完了まで待機し、その結果を返します。イベントが発生した時点で受信するには `stream` を、すぐに戻るには `background` を設定してください。
`GET /api/invocations/<id>`	呼び出しの状態を返し、完了後に出力を返します。
`GET /api/invocations/<id>/events?after=<event-id>`	`<event-id>`の後に続く保存済みのイベントをストリーミングします。このエンドポイントを使ってストリームに再接続してください。

リクエスト本文

POST /api/invocationsのリクエストボディは以下のフィールドを受け入れます。サーバーは他のフィールドを含むリクエストを拒否します。

フィールド	説明
`id`	Required. 各呼び出しごとに生成するUUID。サーバーはIDを冪等キーとして扱い、同じIDで同じリクエストを再送信すると、エージェントを再度実行する代わりに既存の呼び出しを返します。別のリクエストにIDを再利用すると `409` エラーが返されます。
`session_id`	その呼び出しが属する会話です。セッションIDを共有する呼び出しは、順番に1つずつ実行されます。 CLIテンプレートから生成されるエージェントにはこのフィールドが必要です。
`input`	エージェントへのインプットです。 CLIテンプレートから生成されたエージェントは、メッセージのリスト、または `messages` にリストを持つオブジェクトを受け入れます。
`stream`	イベントをサーバー送信イベント (SSE) として受信するには、`true` に設定してください。
`background`	`true` に設定すると、ステータス URL を含む `202` 応答をすぐに返し、その後、結果をポーリングします。

あなたのエージェントのハンドラーがinputの構造を定義します。 CLIテンプレートから生成されたエージェントは、 input がオブジェクトの場合、以下のフィールドを読み込みます。

フィールド	説明
`messages`	会話の話題は、エージェントに転送されます。
`actor`	エージェントが読み書きする長期記憶を持つアイデンティティ。 `actor`をパスしなければ、エージェントはセッションIDを使うため、メモリーは新しいセッションに引き継がれません。 `actor`はアプリケーションのサインイン済みユーザーから設定し、ユーザーが入力したテキストからではなく設定してください。
`model`	この呼び出しに使用するモデル（エージェントコードに設定されたモデルの代わり）
`resume`	ツール呼び出しの承認など、人間の入力のために一時停止したエージェントへの応答。エージェントが一時停止したとき、その呼び出しの `status` は `interrupted` です。続行するには、同じ `session_id` を使用して、新しい呼び出しで `resume` を送信してください。

エージェントがセッション間でユーザーについて学習した内容を想起できるようにするために、ユーザーのIDを actorとして渡してください。

{
  "id": "550e8400-e29b-41d4-a716-446655440000",
  "session_id": "support-case-123",
  "input": {
    "messages": [{ "role": "user", "content": "What does Databricks do?" }],
    "actor": "user-42"
  }
}

エージェント・ブリックス

端末からデプロイ済みエージェントをテストするには、 agentbricks endpoint invokeをご利用ください。コマンドはアプリを特定し、CLIプロファイルを使用して認証します。

agentbricks --profile <profile> endpoint invoke agent-bricks-<name> \
  --path /api/invocations \
  --json "{\"id\":\"$(uuidgen)\",\"session_id\":\"$(uuidgen)\",\"input\":[{\"role\":\"user\",\"content\":\"Hello\"}]}"

レスポンスをストリーミングするには、JSONボディに "stream":true を追加し、--sse を渡してください。 agentbricks dev を使ってローカルで実行しながらエージェントをテストするには、アプリ名を --url http://localhost:8000 に置き換えてください。

REST API

アプリのURLを取得しましょう。出力の URL フィールドは呼び出しAPIの基本URLです。
```
agentbricks --profile <profile> deployments get agent-bricks-<name>
```
プロフィール用にOAuthトークンを取得しましょう。出力は access_token フィールド内のトークンを含みます。
```
databricks auth token --profile <profile>
```

要求を送信します。

curl --request POST \
  --url <app-url>/api/invocations \
  --header 'Authorization: Bearer <OAuth token>' \
  --header 'content-type: application/json' \
  --data '{
    "id": "550e8400-e29b-41d4-a716-446655440000",
    "session_id": "support-case-123",
    "input": [{ "role": "user", "content": "What does Databricks do?" }]
  }'

レスポンスには呼び出し id、その status、そしてエージェントが返す値が記載された output フィールドが含まれています。サーバーは出力スキーマを必要としません。ハンドラはJSONで直列化可能な任意の値を返すことができます。 CLIテンプレートから生成されたエージェントは、以下のフィールドを持つオブジェクトを返します:

output: この呼び出しでエージェントが生成したメッセージ。
status: completed、またはエージェントが人間の入力のために一時停止した場合は interrupted。

Python

次の例では、Databricks SDK を使用してアプリの URL を取得し、OAuth トークンを生成した後、呼び出し API を呼び出しています。 WorkspaceClientはOAuth認証を使用しなければなりません。

import uuid

import requests
from databricks.sdk import WorkspaceClient

w = WorkspaceClient()
app_url = w.apps.get("agent-bricks-<name>").url
session_id = str(uuid.uuid4())

response = requests.post(
    f"{app_url}/api/invocations",
    headers=w.config.authenticate(),
    json={
        "id": str(uuid.uuid4()),
        "session_id": session_id,
        "input": [{"role": "user", "content": "What does Databricks do?"}],
    },
)
response.raise_for_status()
print(response.json()["output"])

会話を続けるには、同じ session_id と新しい idで次のメッセージを送ってください。

ストリーミング、バックグラウンドでの実行、再接続

ストリーム:セット "stream": true。その応答は、 run.started および run.completed (または run.failed)イベントに加え、エージェントが発信するイベント(ストリーミングテキストを含む delta イベントなど)を含むSSEストリームです。各イベントにはIDがあります。
バックグラウンドでの実行: "background": true を設定します。サーバーは status_url を含む 202 応答を返します。 completedをポーリングして、ステータスがGET /api/invocations/<id>になるまで続けてください。 "stream": true も設定すると、応答にはイベントを読み取ることができる events_url が含まれます。
再接続: ストリームが切断された場合は、最後に受信したイベントのIDでGET /api/invocations/<id>/events?after=<event-id>を呼び出してください。

以下のPython例は応答をストリーミングしています:

with requests.post(
    f"{app_url}/api/invocations",
    headers=w.config.authenticate(),
    json={
        "id": str(uuid.uuid4()),
        "session_id": session_id,
        "input": [{"role": "user", "content": "Summarize our last conversation."}],
        "stream": True,
    },
    stream=True,
) as response:
    response.raise_for_status()
    for line in response.iter_lines(decode_unicode=True):
        if line.startswith("data: "):
            print(line[len("data: "):])

複数のインスタンスでエージェントをデプロイする場合は、セッション ID を X-Routing-Key ヘッダーに送信し、セッション内のすべてのリクエストが同じインスタンスにルーティングされるようにします。

レガシーMLflowを使用しているエージェントに問い合わせ `AgentServer`

このセクションでは、レガシーエージェントサーバーを使用して Databricks Apps にデプロイするエージェント (MLflow AgentServer または LongRunningAgentServerと ResponsesAgent インターフェース) について説明します。これらのエージェントは、/responses で OpenAI Responses API を提供しています。

LongRunningAgentServer は同じ API を提供しているため、以下の例はこれにも当てはまります。また、バックグラウンド実行もサポートしています。リクエスト内で background を true に設定し、その後 GET /responses/<response-id>?stream=true&starting_after=<sequence-number> で応答を取得できます。これは、そのシーケンス番号以降のイベントをストリーミングします。

Databricks OpenAIクライアント

Databricksはこれらのエージェントに対して DatabricksのOpenAIクライアントを推奨しています。モデル名に apps/ のプレフィックスを入れてください。

from databricks.sdk import WorkspaceClient
from databricks_openai import DatabricksOpenAI

input_msgs = [{"role": "user", "content": "What does Databricks do?"}]
app_name = "<agent-app-name>"

# The WorkspaceClient must use OAuth authentication.
w = WorkspaceClient()
client = DatabricksOpenAI(workspace_client=w)

# Non-streaming request
response = client.responses.create(model=f"apps/{app_name}", input=input_msgs)
print(response)

# Streaming request
streaming_response = client.responses.create(
    model=f"apps/{app_name}", input=input_msgs, stream=True
)
for chunk in streaming_response:
    print(chunk)

custom_inputsを通すには、extra_bodyパラメータを使用します:

response = client.responses.create(
    model=f"apps/{app_name}",
    input=input_msgs,
    extra_body={"custom_inputs": {"id": 5}},
)

リクエストのトレースIDを取得するには、 x-mlflow-return-trace-id ヘッダーを含めてください。次に、MLflow get_trace を使用して完全なトレースを取得します。

response = client.responses.create(
    model=f"apps/{app_name}",
    input=input_msgs,
    extra_headers={"x-mlflow-return-trace-id": "true"},
)
trace_id = response.metadata["trace_id"]
trace = client.get_trace(trace_id)

REST API

リクエストはアプリのURLの /responses パスに送ります。リクエスト本体はOpenAIレスポンスAPIに準拠しているため、それをサポートするHTTPクライアントやツールを使えます。

curl --request POST \
  --url <app-url>/responses \
  --header 'Authorization: Bearer <OAuth token>' \
  --header 'content-type: application/json' \
  --data '{
    "input": [{ "role": "user", "content": "hi" }],
    "stream": true
  }'

custom_inputsを渡すには、リクエスト本体に追加してください:

curl --request POST \
  --url <app-url>/responses \
  --header 'Authorization: Bearer <OAuth token>' \
  --header 'content-type: application/json' \
  --data '{
    "input": [{ "role": "user", "content": "hi" }],
    "custom_inputs": { "id": 5 }
  }'

トレースIDを取得するには、 x-mlflow-return-trace-id: true ヘッダーを含めてください。レスポンスボディには metadata.trace_id フィールド内にトレースIDが含まれています。ストリーミングリクエストの場合、トレースIDはストリームの終わり近くに別のSSEイベント(data: {"trace_id": "tr-..."})として届きます。

Model Servingでレガシーエージェントを照会

Model Serving エンドポイントにデプロイされたレガシーエージェントには、このセクションを使用してください。認証はDatabricks OAuthトークンか個人アクセストークンで行うことができます。これらのエージェントをDatabricks Appsに移行するには、「 Model ServingからDatabricks Appsへのエージェント移行」をご覧ください。

Databricks OpenAIクライアント

ResponsesAgentインターフェースを使用するエージェントの場合、エンドポイント名をモデルとしてresponses.createを呼び出します。

from databricks_openai import DatabricksOpenAI

input_msgs = [{"role": "user", "content": "What does Databricks do?"}]
endpoint = "<agent-endpoint-name>"

client = DatabricksOpenAI()

# Non-streaming request. Calls predict.
response = client.responses.create(model=endpoint, input=input_msgs)
print(response)

# Streaming request. Calls predict_stream.
streaming_response = client.responses.create(model=endpoint, input=input_msgs, stream=True)
for chunk in streaming_response:
    print(chunk)

レガシー ChatAgent や ChatModel インターフェースを使用しているエージェントは、チャット完了クライアントをご利用ください:

from databricks.sdk import WorkspaceClient

messages = [{"role": "user", "content": "What does Databricks do?"}]
endpoint = "<agent-endpoint-name>"

client = WorkspaceClient().serving_endpoints.get_open_ai_client()
response = client.chat.completions.create(model=endpoint, messages=messages)
print(response)

どちらのクライアントでも、extra_body パラメーターに custom_inputs または databricks_options を渡します。例えば、 extra_body={"databricks_options": {"return_trace": True}} は応答とともにトレースを返します。

REST API

ResponsesAgentインターフェースを使用するエージェントは、エンドポイント名をモデルとして/serving-endpoints/responsesにリクエストを送信します。

curl --request POST \
  --url https://<workspace-url>/serving-endpoints/responses \
  --header 'Authorization: Bearer <token>' \
  --header 'content-type: application/json' \
  --data '{
    "model": "<agent-endpoint-name>",
    "input": [{ "role": "user", "content": "hi" }],
    "stream": true
  }'

ChatAgentまたはChatModelインターフェースを使用するエージェントは、/serving-endpoints/chat/completionsの代わりにmessagesリストを使用して、inputへのリクエストを送信してください。 custom_inputsやdatabricks_optionsを渡すには、それらをリクエスト本体に追加してください。また、エンドポイントの /serving-endpoints/<agent-endpoint-name>/invocations URLにリクエストを送信することも可能です。詳細については、エンドポイントの背後にある個々のモデルを照会するを参照してください。

AI プレイグラウンド

コードを書かずにModel Serservingでエージェントとチャットするには、AI Playground を開き、エージェントのサービングエンドポイントを選択してください。 AI Playgroundからエージェントにcustom_inputsを渡すには、「AI Playground とレビューアプリで custom_inputs を提供する」をご覧ください。

SQL with

ai_queryを使って、SQL から Model Serving のエージェントに問い合わせます。構文とパラメータについては、ai_query関数を参照してください。

SELECT ai_query(
  "<agent-endpoint-name>", question
) FROM (VALUES ('what is MLflow?'), ('how does MLflow work?')) AS t(question);

その他のリソース

エージェントサーバー
エージェントランタイム
生産監視の設定
基盤モデルと埋め込みモデルを問い合わせる: エージェントではなく、基盤モデルやその他のモデルに直接クエリを実行します。

フィードバック

このページはお役に立ちましたか?

最終更新: 2026-10-08

Azure Databricks にデプロイされたクエリ エージェント

DurableAgentServerを使用するエージェントをクエリする

リクエスト本文

エージェント・ブリックス

REST API

Python

ストリーミング、バックグラウンドでの実行、再接続

レガシーMLflowを使用しているエージェントに問い合わせ AgentServer

Databricks OpenAIクライアント

REST API

Model Servingでレガシーエージェントを照会

Databricks OpenAIクライアント

REST API

AI プレイグラウンド

SQL with

その他のリソース

フィードバック

その他のリソース

Azure Databricks にデプロイされたクエリエージェント

`DurableAgentServer`を使用するエージェントをクエリする

レガシーMLflowを使用しているエージェントに問い合わせ `AgentServer`