集成 REST API 与 OpenAPI¶
Supported in ADKPython v0.1.0
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）和请求体 schema。
- 对于每个发现的操作，
-
RestApiTool功能：每个生成的RestApiTool：- Schema 生成：根据操作的参数和请求体动态创建
FunctionDeclaration。该 schema 会告诉 LLM 如何调用该工具（期望哪些参数）。 - 执行：当 LLM 调用该工具时，工具会使用 LLM 给出的参数和 OpenAPI 规范构造 HTTP 请求，包括 URL、请求头、查询参数和请求体。如果配置了身份验证，工具会进行相应处理，并使用
httpx库以 600 秒的读写超时异步执行 API 调用（可通过httpx_client_factory自定义）。
- Schema 生成：根据操作的参数和请求体动态创建
-
身份验证：
- 在初始化
OpenAPIToolset时，你可以配置全局身份验证（如 API Key 或 OAuth - 详情请参阅工具身份验证说明）。这些认证配置将自动应用到该工具集生成的所有RestApiTool实例中。
- 在初始化
使用工作流程¶
请按照以下步骤将 OpenAPI 接口集成到你的智能体中：
- 准备规范：获取你的 OpenAPI 规范文档（例如，从
.json或.yaml文件加载，或从 URL 获取）。 -
实例化工具集：创建
OpenAPIToolset实例，传入规范内容和类型（spec_str/spec_dict、spec_str_type）。如果 API 要求，还需提供身份验证详细信息（auth_scheme、auth_credential）。 -
接入智能体：将该工具集放入
LlmAgent的tools列表中。 -
指示智能体：更新智能体的指令，告知它新增的 API 能力以及它可以使用的工具名称（例如
list_pets、create_pet）。根据规范生成的工具描述同样会有助于 LLM 理解。 - 运行智能体：使用
Runner执行你的智能体。当 LLM 判定需要调用某个 API 时，它会生成一个指向相应RestApiTool的函数调用，随后由该工具自动处理 HTTP 请求。
示例¶
以下示例演示了如何从简单的宠物商店 (Petstore) OpenAPI 规范生成工具，并引导智能体与模拟服务进行交互。
代码：Petstore 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.")
高级配置¶
Note
本节中的参数需要 ADK Python v2.0.0 或更高版本。
使用 header_provider 自定义请求头¶
要为每个 API 请求动态添加自定义请求头，请向 header_provider 提供一个可调用对象。该可调用对象接收 ReadonlyContext 并返回一个请求头字典，如以下示例所示：
from google.adk.agents.readonly_context import ReadonlyContext
from google.adk.tools.openapi_tool.openapi_spec_parser.openapi_toolset import OpenAPIToolset
openapi_spec = {...} # 从你的 OpenAPI 文件加载的。
def my_header_provider(context: ReadonlyContext) -> dict[str, str]:
return {"X-Request-ID": context.invocation_id}
toolset = OpenAPIToolset(
spec_dict=openapi_spec,
header_provider=my_header_provider,
)
使用 ssl_verify 配置 SSL 证书验证¶
要为工具集中的所有工具配置 SSL 证书验证，请使用 ssl_verify 参数。以下示例演示了如何在使用自定义 CA 证书的环境中配置 SSL 验证：
from google.adk.tools.openapi_tool.openapi_spec_parser.openapi_toolset import OpenAPIToolset
openapi_spec = {...} # 从你的 OpenAPI 文件加载的。
toolset = OpenAPIToolset(
spec_dict=openapi_spec,
ssl_verify="/path/to/ca.pem",
)
不要禁用验证
ssl_verify 也接受 True、False 或 ssl.SSLContext。将 ssl_verify=False 会完全禁用证书验证，并使请求暴露于拦截风险。请改用 CA 捆绑包路径或自定义 SSLContext。
使用 tool_name_prefix 设置工具名称前缀¶
要为工具集生成的所有工具名称添加前缀，请使用 tool_name_prefix 参数。以下示例演示了如何在使用多个 OpenAPI 规范时避免名称冲突：
