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

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

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

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

カスタム コンテナをデプロイする場合は、Deployment の仕様の 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"）が必要です。
• Playground: 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)))