OpenAPI 整合¶
使用 OpenAPI 整合 REST API¶
Agent Development Kit (ADK) 透過自動從 OpenAPI Specification (v3.x) 產生可呼叫的工具（tools），大幅簡化與外部 REST API 的互動流程。這讓您無需為每個 API 端點手動定義個別的工具函式。
Core Benefit
使用 OpenAPIToolset 可根據你現有的 API 文件（OpenAPI 規格）即時建立 agent 工具（RestApiTool），讓 agent 能無縫呼叫你的網路服務。
主要元件¶
OpenAPIToolset：這是你主要會使用的類別。你需要以 OpenAPI 規格初始化它，並由它負責解析與產生工具。RestApiTool：這個類別代表單一、可呼叫的 API 操作（例如GET /pets/{petId}或POST /pets）。OpenAPIToolset會針對你規格中定義的每個操作建立一個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，則會根據方法與路徑自動產生名稱。 - 工具描述：會使用操作中的
summary或description作為大型語言模型 (LLM) 的描述。 - API 詳細資訊：會在內部儲存必要的 HTTP 方法、路徑、伺服器基礎 URL、參數（路徑、查詢、標頭、Cookie）以及請求主體 schema。
- 對於每個發現的操作，
-
RestApiTool功能：每個產生的RestApiTool：- Schema 產生：根據操作的參數與請求主體動態建立
FunctionDeclaration。這個 schema 會告訴大型語言模型 (LLM) 如何呼叫該工具（需要哪些參數）。 - 執行：當 LLM 呼叫時，會根據 LLM 提供的參數與 OpenAPI 規格中的細節，組成正確的 HTTP 請求（URL、標頭、查詢參數、主體）。同時會處理驗證（若有設定），並透過
requests函式庫執行 API 呼叫。 - 回應處理：將 API 回應（通常為 JSON）回傳給 agent 流程。
- Schema 產生：根據操作的參數與請求主體動態建立
-
驗證機制：你可以在初始化
OpenAPIToolset時設定全域驗證（例如 API KEY 或 OAuth，詳見 Authentication）。此驗證設定會自動套用至所有產生的RestApiTool實例。
使用流程¶
請依照下列步驟，將 OpenAPI 規格整合進你的 agent：
- 取得規格：取得你的 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 # Example with a JSON string openapi_spec_json = '...' # Your OpenAPI JSON string toolset = OpenAPIToolset(spec_str=openapi_spec_json, spec_str_type="json") # Example with a dictionary # openapi_spec_dict = {...} # Your OpenAPI spec as a dict # toolset = OpenAPIToolset(spec_dict=openapi_spec_dict) -
新增至 agent：將取得的 tools 加入你的
LlmAgent的tools清單中。 -
指示 agent：更新你的 agent 指令，讓其了解新的 API 功能，以及它可以使用的工具名稱（例如：
list_pets、create_pet）。根據 OpenAPI 規格自動產生的工具描述也會協助大型語言模型 (LLM) 理解如何使用這些工具。 - 執行 agent：使用
Runner執行你的 agent。當大型語言模型 (LLM) 判斷需要呼叫某個 API 時，會產生針對適當RestApiTool的函式呼叫，該工具會自動處理 HTTP 請求。
範例¶
本範例展示如何從一個簡單的 Pet Store API OpenAPI 規格產生工具（使用 httpbin.org 來回傳模擬資料），並透過 agent 與這些工具互動。
程式碼：Pet Store 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.")