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 时，该过程包括以下主要步骤：

初始化和解析：
- 你向 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)

获取工具：从工具集获取生成的 RestApiTool 实例列表。

api_tools = toolset.get_tools()
# 或通过生成的名称（snake_case operationId）获取特定工具
# specific_tool = toolset.get_tool("list_pets")

添加到智能体：将获取的工具包含在你的 LlmAgent 的 tools 列表中。

from google.adk.agents import LlmAgent

my_agent = LlmAgent(
    name="api_interacting_agent",
    model="gemini-2.0-flash", # 或你偏好的模型
    tools=api_tools, # 传递生成的工具列表
    # ... 其他智能体配置 ...
)

指导智能体：更新你的智能体指令，告知它新的 API 能力和它可以使用的工具名称（例如，list_pets，create_pet）。从规范生成的工具描述也将帮助 LLM。
运行智能体：使用 Runner , CUI 执行你的智能体。当 LLM 确定它需要调用其中一个 API 时，它会生成一个针对相应 RestApiTool 的函数调用，该工具将自动处理 HTTP 请求。

示例¶

这个示例演示了从简单的宠物商店 OpenAPI 规范生成工具（使用 httpbin.org 进行模拟响应）并通过智能体与它们交互。

???+ "代码：宠物商店 API" {: #code-pet-store-api}

```python title="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.")

```