集成 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时，你可以配置全局身份验证（如 API 密钥或 OAuth - 详情请参阅身份验证）。此身份验证配置将自动应用于所有生成的RestApiTool实例。 -
RestApiTool运行逻辑：- 模式生成：根据操作参数动态创建
FunctionDeclaration，告知 LLM 如何正确调用（需要哪些参数）。 - 执行请求：被 LLM 调用时，它会结合 LLM 提供的参数和规范详情构建正确的 HTTP 请求。它会自动处理认证逻辑（如果已配置），并使用
requests库执行调用。 - 响应处理：将 API 返回的结果（通常为 JSON）传回智能体流程。
- 模式生成：根据操作参数动态创建
-
身份验证：
- 在初始化
OpenAPIToolset时，你可以配置全局身份验证（如 API Key 或 OAuth - 详情请参阅工具身份验证说明）。这些认证配置将自动应用到该工具集生成的所有RestApiTool实例中。
- 在初始化
使用工作流程¶
请按照以下步骤将 OpenAPI 接口集成到你的智能体中：
- 准备规范：获取你的 OpenAPI 规范文档（支持本地加载或通过 URL 获取）。
-
实例化工具集：创建
OpenAPIToolset实例。 -
接入智能体：将该工具集放入
LlmAgent的tools列表中。 -
下达指令：在智能体的系统提词中告知它现在具备了这些 API 能力。
- 执行任务：使用
Runner启动。当模型判断需要调用 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.")