OpenAPI 集成
通过 OpenAPI 集成 REST API
ADK 通过直接从 OpenAPI 规范 (v3.x) 自动生成可调用工具，简化了与外部 REST API 的交互过程。这消除了为每个 API 端点手动定义单独功能工具的需求。
核心优势
使用 OpenAPIToolset 可以立即从现有 API 文档（OpenAPI 规范）创建代理工具（RestApiTool），使代理能够无缝调用您的 Web 服务。
核心组件
OpenAPIToolset：这是您将使用的主要类。通过 OpenAPI 规范初始化后，它会自动处理规范的解析和工具生成。RestApiTool：这个类代表单个可调用的 API 操作（如GET /pets/{petId}或POST /pets）。OpenAPIToolset会为规范中定义的每个操作创建一个RestApiTool实例。
工作原理
当使用 OpenAPIToolset 时，整个过程包含以下主要步骤：
-
初始化与解析：
- 您可以通过 Python 字典、JSON 字符串或 YAML 字符串的形式向
OpenAPIToolset提供 OpenAPI 规范。 - 工具集内部会解析规范，解析所有内部引用（
$ref）以理解完整的 API 结构。
- 您可以通过 Python 字典、JSON 字符串或 YAML 字符串的形式向
-
操作发现：
- 识别规范中定义的所有有效 API 操作（例如
GET、POST、PUT、DELETE）。
- 识别规范中定义的所有有效 API 操作（例如
-
工具生成：
- 对于每个发现的操作，
OpenAPIToolset会自动创建对应的RestApiTool实例。 - 工具名称：从规范中的
operationId派生（转换为snake_case，最多 60 个字符）。如果缺少operationId，则根据方法和路径生成名称。 - 工具描述：使用操作的
summary或description作为大模型的提示词。 - API 详情：内部存储所需的 HTTP 方法、路径、服务器基础 URL、参数（路径、查询、头部、cookie）以及请求体模式。
- 对于每个发现的操作，
-
RestApiTool功能：每个生成的RestApiTool：- 模式生成：根据操作的参数和请求体动态创建
FunctionDeclaration。该模式会告知大模型如何调用工具（预期参数）。 - 执行：当大模型调用时，它会使用大模型提供的参数和 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 = {...} # 您的 OpenAPI 规范字典 # toolset = OpenAPIToolset(spec_dict=openapi_spec_dict) -
获取工具：从工具集中获取生成的
RestApiTool实例列表。 -
添加到代理：将获取的工具包含在
LlmAgent的tools列表中。 -
指导代理：更新代理的指令，告知其新的 API 功能以及可使用的工具名称（例如
list_pets、create_pet）。从规范生成的工具描述也将帮助大模型理解。 - 运行代理：使用
Runner执行代理。当大模型确定需要调用某个 API 时，它会生成针对相应RestApiTool的函数调用，后者会自动处理 HTTP 请求。
示例
本示例演示了如何从简单的 Pet Store OpenAPI 规范生成工具（使用 httpbin.org 模拟响应），并通过代理与之交互。
代码：Pet Store API
openapi_example.py
import asyncio
import uuid # For unique session IDs
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
# --- 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 ---
generated_tools_list = []
try:
# Instantiate the toolset with the spec string
petstore_toolset = OpenAPIToolset(
spec_str=openapi_spec_string,
spec_str_type="json"
# No authentication needed for httpbin.org
)
# Get all tools generated from the spec
generated_tools_list = petstore_toolset.get_tools()
print(f"Generated {len(generated_tools_list)} tools from OpenAPI spec:")
for tool in generated_tools_list:
# Tool names are snake_case versions of operationId
print(f"- Tool Name: '{tool.name}', Description: {tool.description[:60]}...")
except ValueError as ve:
print(f"Validation Error creating OpenAPIToolset: {ve}")
# Handle error appropriately, maybe exit or skip agent creation
except Exception as e:
print(f"Unexpected Error creating OpenAPIToolset: {e}")
# Handle error appropriately
# --- Agent Definition ---
openapi_agent = LlmAgent(
name=AGENT_NAME_OPENAPI,
model=GEMINI_MODEL,
tools=generated_tools_list, # Pass the list of RestApiTool objects
instruction=f"""You are a Pet Store assistant managing pets via an API.
Use the available tools to fulfill user requests.
Available tools: {', '.join([t.name for t in generated_tools_list])}.
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 ---
session_service_openapi = InMemorySessionService()
runner_openapi = Runner(
agent=openapi_agent, app_name=APP_NAME_OPENAPI, session_service=session_service_openapi
)
session_openapi = session_service_openapi.create_session(
app_name=APP_NAME_OPENAPI, user_id=USER_ID_OPENAPI, session_id=SESSION_ID_OPENAPI
)
# --- Agent Interaction Function ---
async def call_openapi_agent_async(query):
print("\n--- Running OpenAPI Pet Store Agent ---")
print(f"Query: {query}")
if not generated_tools_list:
print("Skipping execution: No tools were generated.")
print("-" * 30)
return
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():
# Trigger listPets
await call_openapi_agent_async("Show me the pets available.")
# Trigger createPet
await call_openapi_agent_async("Please add a new dog named 'Dukey'.")
# Trigger showPetById
await call_openapi_agent_async("Get info for pet with ID 123.")
# --- 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.")