Agent Platform ランタイムの契約

Agent Runtime は、アプリケーション フレームワークに依存しないように設計されています。カスタム コンテナまたは Dockerfile を使用してエージェントをデプロイする場合は、クエリを正常に処理するために、コンテナがランタイム契約に準拠している必要があります。

デプロイ方法の詳細については、エージェントをデプロイするをご覧ください。

制約と要件

Agent Runtime にカスタム コンテナをデプロイするには、コンテナがポート 8080 の 0.0.0.0 で HTTP リクエストをリッスンする必要があります。

エンドポイント（省略可）

コンテナは任意のカスタム HTTP エンドポイントを公開できます。これらのカスタム エンドポイントを呼び出すには、デプロイされたエージェントの基盤となる API にリクエストを送信します。詳細については、基盤となる API を介してデプロイされたエージェントを使用するをご覧ください。

これらのエンドポイントは API レベルでは省略可能ですが、実装すると次の統合機能が有効になります。

• Python SDK のサポート: /api/reasoning_engine と /api/stream_reasoning_engine の両方を実装すると、Agent Platform Python SDK を介してデプロイされたエージェントを使用できます。
• プレイグラウンドのサポート:Google Cloud コンソール プレイグラウンドでエージェントを操作する場合は、/api/stream_reasoning_engine を実装する必要があります。詳細については、プレイグラウンドのサポートをご覧ください。

これらの機能を使用する場合は、次のエンドポイントを実装する必要があります。

• /api/reasoning_engine: reasoningEngines/query REST API または Python SDK の同期クエリ メソッドと非同期クエリ メソッドに送信されたクエリのサービングに使用されます。
• /api/stream_reasoning_engine: reasoningEngines/streamQuery REST API または Python SDK のストリーミング クエリ メソッドに送信されたクエリのサービングに使用されます。

クラスメソッドと実行モード

カスタム コンテナをデプロイする場合は、デプロイ仕様の classMethods リストでサポートされているクラスメソッドを宣言する必要があります。これらのクラスメソッドは、エージェントの開発時に定義したオペレーションに対応しています（カスタム メソッドの登録とサポートされているオペレーションを使用してエージェントをクエリするをご覧ください）。宣言された各メソッドには、ルーティング方法を決定する name と api_mode があります。

API モード	実行タイプ	ルーティング エンドポイント
""（空の文字列）または "async"	単項（リクエスト-レスポンス）	/api/reasoning_engine
"stream" または "async_stream"	ストリーミング	/api/stream_reasoning_engine

クライアントがメソッドを呼び出すと、Agent Runtime サービスはコンテナ内の対応するルーティング エンドポイントに POST リクエストを送信します。リクエストの JSON 本文には、class_method フィールド（メソッド名と一致）と input フィールドが含まれています。

統合に必要なメソッド

Python SDK または Google Cloud コンソール プレイグラウンドを使用するには、コンテナでこれらの統合が想定する特定のメソッドを実装する必要があります。

• SDK 標準クエリ: query（モード "" または "async"）と stream_query（モード "stream" または "async_stream"）が必要です。
• プレイグラウンド: stream_query（モード "stream" または "async_stream"）が必要です。

ADK の統合

Agent Development Kit（ADK）で構築されたエージェントをデプロイする場合は、任意のプログラミング言語とサーバー フレームワークで独自のプロキシ コンテナを構築できます。ADK のすべての機能をサポートするには、コンテナで ADK 契約で定義されたメソッドを実装する必要があります。詳細については、ADK エージェントを使用すると ADK エージェントを登録して管理するをご覧ください。

AdkApp テンプレートをリファレンス実装として参照できます。詳細については、AdkApp リファレンス ドキュメントと AdkApp ソースコードをご覧ください。

ADK のバージョンを更新するときに Google が自動的に更新を処理するようにするには、独自の API サーバーを作成するのではなく、ADK が提供するエージェントのデプロイ用ツールを使用する必要があります。詳細については、ADK のデプロイ ドキュメントをご覧ください。

API 仕様

/api/reasoning_engine と /api/stream_reasoning_engine の両方が、次のフィールドを含む JSON 本文を含む HTTP POST リクエストを受け取ります。

• class_method（文字列）: 呼び出す基盤となるエージェントのメソッド名（query、stream_query など）。
• input（JSON オブジェクト）: 指定されたクラスメソッドに渡される引数。

/api/reasoning_engine（単項）

• リクエスト メソッド: POST
• リクエストの本文: json { "class_method": "query", "input": { "message": "What is the capital of France?" } }
• レスポンス: エージェント実行の出力を含む JSON オブジェクト。json { "output": "The capital of France is Paris." }

/api/stream_reasoning_engine（ストリーミング）

• リクエスト メソッド: POST
• リクエストの本文: json { "class_method": "stream_query", "input": { "message": "Tell me a short story." } }
• レスポンス: 行区切りの JSON（ndjson）のストリーム。各行は、レスポンスの JSON エンコードされたチャンクです。 json {"output": "Once"} {"output": " upon"} {"output": " a time..."}

API サーバーの例（Python）

次の例は、Agent Platform ランタイム コントラクトを実装する Python の FastAPI サーバーです。このサーバーは、エージェントの例（SimpleAgent）をラップし、ルーティングとエンコードを処理します。SimpleAgent は、独自のエージェント実装に置き換えることができます。

この例を実行するには、fastapi、uvicorn、pydantic がインストールされていることを確認してください。

import inspect
import json
import logging
import os
import uvicorn
from fastapi import FastAPI, encoders, responses
from pydantic import BaseModel

app = FastAPI()

# Define the request body structure
class QueryRequest(BaseModel):
    input: dict | None = None
    class_method: str

# Example Agent implementation
class SimpleAgent:
    def query(self, message: str) -> str:
        return f"Echo: {message}"

    async def stream_query(self, message: str):
        words = message.split()
        for word in words:
            yield {"output": word + " "}

agent = SimpleAgent()

def _encode_chunk_to_json(chunk):
    """Encodes a chunk to a JSON string with a newline."""
    try:
        json_chunk = encoders.jsonable_encoder(chunk)
        return json.dumps(json_chunk) + "\n"
    except Exception:
        logging.exception("Failed to encode chunk")
        return None

async def json_generator(output):
    async for chunk in output:
        encoded_chunk = _encode_chunk_to_json(chunk)
        if encoded_chunk is None:
            break
        yield encoded_chunk

async def _invoke_callable_or_raise(invocation_callable, invocation_payload):
    if inspect.iscoroutinefunction(invocation_callable):
        return await invocation_callable(**invocation_payload)
    else:
        return invocation_callable(**invocation_payload)

@app.post("/api/reasoning_engine")
async def query_endpoint(request: QueryRequest) -> responses.JSONResponse:
    try:
        method = getattr(agent, request.class_method)
    except AttributeError:
        return responses.JSONResponse(
            status_code=400,
            content={"error": f"Method {request.class_method} not found on agent"}
        )

    output = await _invoke_callable_or_raise(method, request.input or {})

    try:
        json_serialized_content = encoders.jsonable_encoder({"output": output})
    except ValueError as encoding_error:
        logging.exception("Failed to JSON-encode response: %s", encoding_error)
        raise encoding_error
    return responses.JSONResponse(content=json_serialized_content)

@app.post("/api/stream_reasoning_engine")
async def stream_query_endpoint(request: QueryRequest) -> responses.StreamingResponse:
    try:
        method = getattr(agent, request.class_method)
    except AttributeError:
        return responses.StreamingResponse(
            content=iter([json.dumps({"error": f"Method {request.class_method} not found"})]),
            status_code=400,
            media_type="application/json"
        )

    output = await _invoke_callable_or_raise(method, request.input or {})
    return responses.StreamingResponse(
        content=json_generator(output),
        media_type="application/json",
    )

if __name__ == "__main__":
    # The container must listen on 0.0.0.0 and port 8080
    uvicorn.run(app, host="0.0.0.0", port=int(os.environ.get("PORT", 8080)))