OpenAPI 통합¶
OpenAPI를 사용한 REST API 통합¶
ADK는 OpenAPI 사양 (v3.x)에서 직접 호출 가능한 도구를 자동으로 생성하여 외부 REST API와의 상호작용을 단순화합니다. 이를 통해 각 API 엔드포인트에 대한 개별 함수 도구를 수동으로 정의할 필요가 없습니다.
핵심 이점
OpenAPIToolset을 사용하여 기존 API 문서(OpenAPI 사양)에서 에이전트 도구(RestApiTool)를 즉시 생성하고, 에이전트가 웹 서비스를 원활하게 호출할 수 있도록 하세요.
주요 구성 요소¶
OpenAPIToolset: 주로 사용하게 될 기본 클래스입니다. OpenAPI 사양으로 초기화하면 도구의 파싱과 생성을 처리합니다.RestApiTool:GET /pets/{petId}또는POST /pets와 같이 호출 가능한 단일 API 작업을 나타내는 클래스입니다.OpenAPIToolset은 사양에 정의된 각 작업에 대해 하나의RestApiTool인스턴스를 생성합니다.
작동 방식¶
OpenAPIToolset을 사용할 때의 과정은 다음과 같은 주요 단계를 포함합니다:
-
초기화 및 파싱:
- OpenAPI 사양을 Python 딕셔너리, JSON 문자열 또는 YAML 문자열로
OpenAPIToolset에 제공합니다. - 도구 세트는 내부적으로 사양을 파싱하여 내부 참조(
$ref)를 확인하고 완전한 API 구조를 이해합니다.
- OpenAPI 사양을 Python 딕셔너리, JSON 문자열 또는 YAML 문자열로
-
작업 검색:
- 사양의
paths객체 내에 정의된 모든 유효한 API 작업(예:GET,POST,PUT,DELETE)을 식별합니다.
- 사양의
-
도구 생성:
- 발견된 각 작업에 대해
OpenAPIToolset은 해당RestApiTool인스턴스를 자동으로 생성합니다. - 도구 이름: 사양의
operationId에서 파생됩니다(snake_case로 변환, 최대 60자).operationId가 없는 경우 메서드와 경로에서 이름이 생성됩니다. - 도구 설명: 작업의
summary또는description을 LLM에 사용합니다. - API 세부 정보: 필요한 HTTP 메서드, 경로, 서버 기본 URL, 매개변수(경로, 쿼리, 헤더, 쿠키) 및 요청 본문 스키마를 내부에 저장합니다.
- 발견된 각 작업에 대해
-
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 사양을 dict로 # toolset = OpenAPIToolset(spec_dict=openapi_spec_dict) -
에이전트에 추가: 검색된 도구를
LlmAgent의tools목록에 포함시킵니다. -
에이전트 지시: 새로운 API 기능과 사용할 수 있는 도구의 이름(예:
list_pets,create_pet)을 알려주도록 에이전트의 지침을 업데이트합니다. 사양에서 생성된 도구 설명도 LLM에 도움이 될 것입니다. - 에이전트 실행:
Runner를 사용하여 에이전트를 실행합니다. LLM이 API 중 하나를 호출해야 한다고 판단하면, 적절한RestApiTool을 대상으로 하는 함수 호출을 생성하고, 그러면 해당 도구가 자동으로 HTTP 요청을 처리합니다.
예제¶
이 예제는 간단한 Pet Store OpenAPI 사양(모의 응답을 위해 httpbin.org 사용)에서 도구를 생성하고 에이전트를 통해 상호 작용하는 방법을 보여줍니다.
코드: 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.")