Integra APIs REST con OpenAPI¶

Compatible con ADKPython v0.1.0

ADK simplifica la interacción con APIs REST externas generando automáticamente herramientas invocables directamente desde una Especificación OpenAPI (v3.x). Esto elimina la necesidad de definir manualmente herramientas de función individuales para cada endpoint de la API.

Beneficio Principal

Usa OpenAPIToolset para crear instantáneamente herramientas de agente (RestApiTool) desde tu documentación de API existente (especificación OpenAPI), permitiendo que los agentes llamen sin problemas a tus servicios web.

Componentes Clave¶

OpenAPIToolset: Esta es la clase principal que usarás. La inicializas con tu especificación OpenAPI, y maneja el análisis y generación de herramientas.
RestApiTool: Esta clase representa una única operación de API invocable (como GET /pets/{petId} o POST /pets). OpenAPIToolset crea una instancia de RestApiTool por cada operación definida en tu especificación.

Cómo Funciona¶

El proceso involucra estos pasos principales cuando usas OpenAPIToolset:

Inicialización y Análisis:
- Proporcionas la especificación OpenAPI a OpenAPIToolset ya sea como un diccionario de Python, un string JSON, o un string YAML.
- El conjunto de herramientas analiza internamente la especificación, resolviendo cualquier referencia interna ($ref) para entender la estructura completa de la API.
Descubrimiento de Operaciones:
- Identifica todas las operaciones de API válidas (ej., GET, POST, PUT, DELETE) definidas dentro del objeto paths de tu especificación.
Generación de Herramientas:
- Por cada operación descubierta, OpenAPIToolset crea automáticamente una instancia correspondiente de RestApiTool.
- Nombre de Herramienta: Derivado del operationId en la especificación (convertido a snake_case, máximo 60 caracteres). Si falta operationId, se genera un nombre desde el método y la ruta.
- Descripción de Herramienta: Usa el summary o description de la operación para el LLM.
- Detalles de API: Almacena internamente el método HTTP requerido, ruta, URL base del servidor, parámetros (path, query, header, cookie), y esquema del cuerpo de la petición.
Funcionalidad de RestApiTool: Cada RestApiTool generado:
- Generación de Esquema: Crea dinámicamente una FunctionDeclaration basada en los parámetros de la operación y el cuerpo de la petición. Este esquema le dice al LLM cómo llamar la herramienta (qué argumentos se esperan).
- Ejecución: Cuando es llamado por el LLM, construye la petición HTTP correcta (URL, headers, parámetros de query, body) usando los argumentos proporcionados por el LLM y los detalles de la especificación OpenAPI. Maneja la autenticación (si está configurada) y ejecuta la llamada a la API usando la librería requests.
- Manejo de Respuesta: Devuelve la respuesta de la API (típicamente JSON) de vuelta al flujo del agente.
Autenticación: Puedes configurar autenticación global (como claves API u OAuth - ver Authentication para detalles) al inicializar OpenAPIToolset. Esta configuración de autenticación se aplica automáticamente a todas las instancias de RestApiTool generadas.

Flujo de Trabajo de Uso¶

Sigue estos pasos para integrar una especificación OpenAPI en tu agente:

Obtener Especificación: Obtén tu documento de especificación OpenAPI (ej., cargar desde un archivo .json o .yaml, obtener desde una URL).

Instanciar Conjunto de Herramientas: Crea una instancia de OpenAPIToolset, pasando el contenido de la especificación y el tipo (spec_str/spec_dict, spec_str_type). Proporciona detalles de autenticación (auth_scheme, auth_credential) si la API lo requiere.

from google.adk.tools.openapi_tool.openapi_spec_parser.openapi_toolset import OpenAPIToolset

# Ejemplo con un string JSON
openapi_spec_json = '...' # Tu string JSON de OpenAPI
toolset = OpenAPIToolset(spec_str=openapi_spec_json, spec_str_type="json")

# Ejemplo con un diccionario
# openapi_spec_dict = {...} # Tu especificación OpenAPI como dict
# toolset = OpenAPIToolset(spec_dict=openapi_spec_dict)

Agregar al Agente: Incluye las herramientas recuperadas en la lista tools de tu LlmAgent.

from google.adk.agents import LlmAgent

my_agent = LlmAgent(
    name="api_interacting_agent",
    model="gemini-2.0-flash", # O tu modelo preferido
    tools=[toolset], # Pasa el conjunto de herramientas
    # ... otra configuración del agente ...
)

Instruir al Agente: Actualiza las instrucciones de tu agente para informarle sobre las nuevas capacidades de API y los nombres de las herramientas que puede usar (ej., list_pets, create_pet). Las descripciones de herramientas generadas desde la especificación también ayudarán al LLM.
Ejecutar Agente: Ejecuta tu agente usando el Runner. Cuando el LLM determine que necesita llamar a una de las APIs, generará una llamada de función apuntando a la RestApiTool apropiada, que luego manejará la petición HTTP automáticamente.

Ejemplo¶

Este ejemplo demuestra la generación de herramientas desde una especificación OpenAPI simple de Pet Store (usando httpbin.org para respuestas simuladas) e interactuando con ellas a través de un agente.

Code: Pet Store 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.")