OpenAPI 集成¶
集成 REST API 与 OpenAPI¶
ADK 通过直接从 OpenAPI 规范 (v3.x) 自动生成可调用的工具，简化了与外部 REST API 的交互。这消除了为每个 API 端点手动定义单独函数工具的需要。
核心优势
使用 OpenAPIToolset 从你现有的 API 文档（OpenAPI 规范）即时创建智能体工具（RestApiTool），使智能体能够无缝调用你的 Web 服务。
关键组件¶
OpenAPIToolset：这是你将使用的主要类。你使用 OpenAPI 规范初始化它，它处理工具的解析和生成。RestApiTool：这个类表示单个可调用的 API 操作（如GET /pets/{petId}或POST /pets）。OpenAPIToolset为规范中定义的每个操作创建一个RestApiTool实例。
工作原理¶
当你使用 OpenAPIToolset 时，该过程包括以下主要步骤：
-
初始化和解析：
- 你向
OpenAPIToolset提供 OpenAPI 规范，可以是 Python 字典、JSON 字符串或 YAML 字符串。 - 工具集在内部解析规范，解析任何内部引用（
$ref）以理解完整的 API 结构。
- 你向
-
操作发现：
- 它识别规范的
paths对象中定义的所有有效 API 操作（例如，GET、POST、PUT、DELETE）。
- 它识别规范的
-
工具生成：
- 对于每个发现的操作，
OpenAPIToolset自动创建相应的RestApiTool实例。 - 工具名称：从规范中的
operationId派生（转换为snake_case，最多 60 个字符）。如果缺少operationId，则从方法和路径生成名称。 - 工具描述：使用操作的
summary或description提供给 LLM。 - API 详情：在内部存储所需的 HTTP 方法、路径、服务器基本 URL、参数（路径、查询、头部、cookie）和请求体模式。
- 对于每个发现的操作，
-
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 = {...} # 你的 OpenAPI 规范作为字典 # toolset = OpenAPIToolset(spec_dict=openapi_spec_dict) -
Add to Agent: Include the retrieved tools in your
LlmAgent'stoolslist. -
Instruct Agent: Update your agent's instructions to inform it about the new API capabilities and the names of the tools it can use (e.g.,
list_pets,create_pet). The tool descriptions generated from the spec will also help the LLM. - Run Agent: Execute your agent using the
Runner. When the LLM determines it needs to call one of the APIs, it will generate a function call targeting the appropriateRestApiTool, which will then handle the HTTP request automatically.
示例¶
这个示例演示了从简单的宠物商店 OpenAPI 规范生成工具（使用 httpbin.org 进行模拟响应）并通过智能体与它们交互。
???+ "代码：宠物商店 API" {: #code-pet-store-api}
```python title="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.")
```