集成 REST API 与 OpenAPI

Supported in ADKPython v0.1.0

ADK 支持直接从 OpenAPI 规范 (v3.x) 自动生成可调用的工具，从而极大地简化了与外部 REST API 的交互。这消除了为每个 API 端点手动定义单独函数工具的繁琐工作。

核心优势

使用 OpenAPIToolset 即可从你现有的 API 文档（OpenAPI 规范）即时创建智能体工具（型为 RestApiTool），使智能体能够无缝调用你的 Web 服务。

Key components

关键组件

How it works

---

工作原理

当你使用 OpenAPIToolset 时，其内部处理流程如下：

1. 初始化与解析：

   • 你向 OpenAPIToolset 提供 OpenAPI 规范（支持 Python 字典、JSON 字符串或 YAML 字符串）。
   • 工具集在内部解析规范，并处理所有的内部引用（$ref），以构建完整的 API 映射。

2. RestApiTool Functionality: Each generated RestApiTool:

   • Schema Generation: Dynamically creates a FunctionDeclaration based on the operation's parameters and request body. This schema tells the LLM how to call the tool (what arguments are expected).
   • Execution: When the LLM calls the tool, the tool constructs the HTTP request, including the URL, headers, query parameters, and body, using the LLM's arguments and the OpenAPI specification. The tool handles authentication if configured, and executes the API call asynchronously using the httpx library.
   • Response Handling: Returns the API response (typically JSON) back to the agent flow.

3. Authentication: You can configure global authentication (like API keys or OAuth - see Authentication for details) when initializing OpenAPIToolset. This authentication configuration is automatically applied to all generated RestApiTool instances.

Usage workflow

1. 身份验证：
   • 在初始化 OpenAPIToolset 时，你可以配置全局身份验证（如 API Key 或 OAuth - 详情请参阅工具身份验证说明）。这些认证配置将自动应用到该工具集生成的所有 RestApiTool 实例中。

---

使用工作流程

请按照以下步骤将 OpenAPI 接口集成到你的智能体中：

1. 准备规范：获取你的 OpenAPI 规范文档（支持本地加载或通过 URL 获取）。

2. 实例化工具集：创建 OpenAPIToolset 实例。

   from google.adk.tools.openapi_tool.openapi_spec_parser.openapi_toolset import OpenAPIToolset
   
   # 方式 A：使用 JSON 字符串
   openapi_spec_json = '...' 
   toolset = OpenAPIToolset(spec_str=openapi_spec_json, spec_str_type="json")
   
   # 方式 B：使用字典
   # toolset = OpenAPIToolset(spec_dict=my_parsed_dict)
   

3. 接入智能体：将该工具集放入 LlmAgent 的 tools 列表中。

   from google.adk.agents import LlmAgent
   
   my_agent = LlmAgent(
       name="api_interacting_agent",
       model="gemini-flash-latest", # Or your preferred model
       tools=[toolset], # Pass the toolset
       # ... other agent config ...
   )
   

4. 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.

5. 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 appropriate RestApiTool, which will then handle the HTTP request automatically.

See it in action

示例

以下示例演示了如何从简单的宠物商店 (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.")