OpenAPI連携¶
OpenAPIによるREST APIの連携¶
ADKは、OpenAPI Specification (v3.x)から直接呼び出し可能なツールを自動的に生成することで、外部REST APIとの対話を簡素化します。これにより、各APIエンドポイントに対して個別の関数ツールを手動で定義する必要がなくなります。
主な利点
OpenAPIToolsetを使用すると、既存のAPIドキュメント（OpenAPI仕様）からエージェントツール（RestApiTool）を即座に作成でき、エージェントがWebサービスをシームレスに呼び出すことが可能になります。
主要コンポーネント¶
OpenAPIToolset: これが主に使用するクラスです。OpenAPI仕様で初期化すると、ツールの解析と生成を処理します。RestApiTool: このクラスは、単一の呼び出し可能なAPI操作（例:GET /pets/{petId}やPOST /pets）を表します。OpenAPIToolsetは、仕様で定義された各操作に対して1つのRestApiToolインスタンスを作成します。
仕組み¶
OpenAPIToolsetを使用する場合、プロセスは主に次のステップを含みます。
-
初期化と解析:
- OpenAPI仕様をPythonの辞書、JSON文字列、またはYAML文字列として
OpenAPIToolsetに提供します。 - ツールセットは内部で仕様を解析し、内部参照（
$ref）を解決して完全なAPI構造を理解します。
- OpenAPI仕様をPythonの辞書、JSON文字列、またはYAML文字列として
-
操作の発見:
- 仕様の
pathsオブジェクト内で定義されたすべての有効なAPI操作（例:GET,POST,PUT,DELETE）を識別します。
- 仕様の
-
ツールの生成:
- 発見された各操作に対して、
OpenAPIToolsetは対応するRestApiToolインスタンスを自動的に作成します。 - ツール名: 仕様の
operationIdから派生します（snake_caseに変換、最大60文字）。operationIdがない場合は、メソッドとパスから名前が生成されます。 - ツールの説明: LLMのために、操作の
summaryまたはdescriptionを使用します。 - API詳細: 必要なHTTPメソッド、パス、サーバーのベースURL、パラメータ（パス、クエリ、ヘッダー、クッキー）、およびリクエストボディのスキーマを内部に保存します。
- 発見された各操作に対して、
-
RestApiToolの機能: 生成された各RestApiToolは次のようになります。- スキーマ生成: 操作のパラメータとリクエストボディに基づいて
FunctionDeclarationを動的に作成します。このスキーマは、LLMにツールの呼び出し方（どの引数が期待されるか）を伝えます。 - 実行: LLMによって呼び出されると、LLMから提供された引数とOpenAPI仕様の詳細を使用して、正しいHTTPリクエスト（URL、ヘッダー、クエリパラメータ、ボディ）を構築します。認証を（設定されていれば）処理し、
requestsライブラリを使用してAPI呼び出しを実行します。 - レスポンス処理: APIレスポンス（通常はJSON）をエージェントのフローに返します。
- スキーマ生成: 操作のパラメータとリクエストボディに基づいて
-
認証:
OpenAPIToolsetを初期化する際に、グローバルな認証（APIキーやOAuthなど - 詳細は認証を参照）を設定できます。この認証設定は、生成されたすべてのRestApiToolインスタンスに自動的に適用されます。
利用ワークフロー¶
OpenAPI仕様をエージェントに統合するには、次の手順に従います。
- 仕様の取得: OpenAPI仕様ドキュメントを取得します（例:
.jsonや.yamlファイルから読み込む、URLから取得する）。 -
ツールセットのインスタンス化:
OpenAPIToolsetインスタンスを作成し、仕様のコンテンツとタイプ（spec_str/spec_dict,spec_str_type）を渡します。APIで必要な場合は、認証情報（auth_scheme,auth_credential）を提供します。from google.adk.tools.openapi_tool.openapi_spec_parser.openapi_toolset import OpenAPIToolset # JSON文字列の例 openapi_spec_json = '...' # あなたのOpenAPI JSON文字列 toolset = OpenAPIToolset(spec_str=openapi_spec_json, spec_str_type="json") # 辞書の例 # openapi_spec_dict = {...} # dictとしてのあなたのOpenAPI仕様 # toolset = OpenAPIToolset(spec_dict=openapi_spec_dict) -
エージェントに追加: 取得したツールを
LlmAgentのtoolsリストに含めます。 -
エージェントへの指示: エージェントの指示を更新し、新しいAPIの能力と使用できるツールの名前（例:
list_pets,create_pet）を伝えます。仕様から生成されたツールの説明もLLMの助けになります。 - エージェントの実行:
Runnerを使用してエージェントを実行します。LLMがいずれかのAPIを呼び出す必要があると判断すると、適切なRestApiToolをターゲットとする関数呼び出しを生成し、それが自動的にHTTPリクエストを処理します。
例¶
この例は、簡単なペットストアのOpenAPI仕様からツールを生成し（モックレスポンスにhttpbin.orgを使用）、エージェントを介してそれらと対話する方法を示しています。
コード: ペットストアAPI
openapi_example.py
# Copyright 2025 Google LLC
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
import asyncio
import uuid # For unique session IDs
from dotenv import load_dotenv
from google.adk.agents import LlmAgent
from google.adk.runners import Runner
from google.adk.sessions import InMemorySessionService
from google.genai import types
# --- OpenAPI Tool Imports ---
from google.adk.tools.openapi_tool.openapi_spec_parser.openapi_toolset import OpenAPIToolset
# --- Load Environment Variables (If ADK tools need them, e.g., API keys) ---
load_dotenv() # Create a .env file in the same directory if needed
# --- Constants ---
APP_NAME_OPENAPI = "openapi_petstore_app"
USER_ID_OPENAPI = "user_openapi_1"
SESSION_ID_OPENAPI = f"session_openapi_{uuid.uuid4()}" # Unique session ID
AGENT_NAME_OPENAPI = "petstore_manager_agent"
GEMINI_MODEL = "gemini-2.0-flash"
# --- Sample OpenAPI Specification (JSON String) ---
# A basic Pet Store API example using httpbin.org as a mock server
openapi_spec_string = """
{
"openapi": "3.0.0",
"info": {
"title": "Simple Pet Store API (Mock)",
"version": "1.0.1",
"description": "An API to manage pets in a store, using httpbin for responses."
},
"servers": [
{
"url": "https://httpbin.org",
"description": "Mock server (httpbin.org)"
}
],
"paths": {
"/get": {
"get": {
"summary": "List all pets (Simulated)",
"operationId": "listPets",
"description": "Simulates returning a list of pets. Uses httpbin's /get endpoint which echoes query parameters.",
"parameters": [
{
"name": "limit",
"in": "query",
"description": "Maximum number of pets to return",
"required": false,
"schema": { "type": "integer", "format": "int32" }
},
{
"name": "status",
"in": "query",
"description": "Filter pets by status",
"required": false,
"schema": { "type": "string", "enum": ["available", "pending", "sold"] }
}
],
"responses": {
"200": {
"description": "A list of pets (echoed query params).",
"content": { "application/json": { "schema": { "type": "object" } } }
}
}
}
},
"/post": {
"post": {
"summary": "Create a pet (Simulated)",
"operationId": "createPet",
"description": "Simulates adding a new pet. Uses httpbin's /post endpoint which echoes the request body.",
"requestBody": {
"description": "Pet object to add",
"required": true,
"content": {
"application/json": {
"schema": {
"type": "object",
"required": ["name"],
"properties": {
"name": {"type": "string", "description": "Name of the pet"},
"tag": {"type": "string", "description": "Optional tag for the pet"}
}
}
}
}
},
"responses": {
"201": {
"description": "Pet created successfully (echoed request body).",
"content": { "application/json": { "schema": { "type": "object" } } }
}
}
}
},
"/get?petId={petId}": {
"get": {
"summary": "Info for a specific pet (Simulated)",
"operationId": "showPetById",
"description": "Simulates returning info for a pet ID. Uses httpbin's /get endpoint.",
"parameters": [
{
"name": "petId",
"in": "path",
"description": "This is actually passed as a query param to httpbin /get",
"required": true,
"schema": { "type": "integer", "format": "int64" }
}
],
"responses": {
"200": {
"description": "Information about the pet (echoed query params)",
"content": { "application/json": { "schema": { "type": "object" } } }
},
"404": { "description": "Pet not found (simulated)" }
}
}
}
}
}
"""
# --- Create OpenAPIToolset ---
petstore_toolset = OpenAPIToolset(
spec_str=openapi_spec_string,
spec_str_type='json',
# No authentication needed for httpbin.org
)
# --- Agent Definition ---
root_agent = LlmAgent(
name=AGENT_NAME_OPENAPI,
model=GEMINI_MODEL,
tools=[petstore_toolset], # Pass the list of RestApiTool objects
instruction="""You are a Pet Store assistant managing pets via an API.
Use the available tools to fulfill user requests.
When creating a pet, confirm the details echoed back by the API.
When listing pets, mention any filters used (like limit or status).
When showing a pet by ID, state the ID you requested.
""",
description="Manages a Pet Store using tools generated from an OpenAPI spec."
)
# --- Session and Runner Setup ---
async def setup_session_and_runner():
session_service_openapi = InMemorySessionService()
runner_openapi = Runner(
agent=root_agent,
app_name=APP_NAME_OPENAPI,
session_service=session_service_openapi,
)
await session_service_openapi.create_session(
app_name=APP_NAME_OPENAPI,
user_id=USER_ID_OPENAPI,
session_id=SESSION_ID_OPENAPI,
)
return runner_openapi
# --- Agent Interaction Function ---
async def call_openapi_agent_async(query, runner_openapi):
print("\n--- Running OpenAPI Pet Store Agent ---")
print(f"Query: {query}")
content = types.Content(role='user', parts=[types.Part(text=query)])
final_response_text = "Agent did not provide a final text response."
try:
async for event in runner_openapi.run_async(
user_id=USER_ID_OPENAPI, session_id=SESSION_ID_OPENAPI, new_message=content
):
# Optional: Detailed event logging for debugging
# print(f" DEBUG Event: Author={event.author}, Type={'Final' if event.is_final_response() else 'Intermediate'}, Content={str(event.content)[:100]}...")
if event.get_function_calls():
call = event.get_function_calls()[0]
print(f" Agent Action: Called function '{call.name}' with args {call.args}")
elif event.get_function_responses():
response = event.get_function_responses()[0]
print(f" Agent Action: Received response for '{response.name}'")
# print(f" Tool Response Snippet: {str(response.response)[:200]}...") # Uncomment for response details
elif event.is_final_response() and event.content and event.content.parts:
# Capture the last final text response
final_response_text = event.content.parts[0].text.strip()
print(f"Agent Final Response: {final_response_text}")
except Exception as e:
print(f"An error occurred during agent run: {e}")
import traceback
traceback.print_exc() # Print full traceback for errors
print("-" * 30)
# --- Run Examples ---
async def run_openapi_example():
runner_openapi = await setup_session_and_runner()
# Trigger listPets
await call_openapi_agent_async("Show me the pets available.", runner_openapi)
# Trigger createPet
await call_openapi_agent_async("Please add a new dog named 'Dukey'.", runner_openapi)
# Trigger showPetById
await call_openapi_agent_async("Get info for pet with ID 123.", runner_openapi)
# --- Execute ---
if __name__ == "__main__":
print("Executing OpenAPI example...")
# Use asyncio.run() for top-level execution
try:
asyncio.run(run_openapi_example())
except RuntimeError as e:
if "cannot be called from a running event loop" in str(e):
print("Info: Cannot run asyncio.run from a running event loop (e.g., Jupyter/Colab).")
# If in Jupyter/Colab, you might need to run like this:
# await run_openapi_example()
else:
raise e
print("OpenAPI example finished.")