OpenAPI를 사용하여 REST API 통합하기¶
ADK에서 지원Python v0.1.0
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가 없는 경우, 메서드와 경로를 기반으로 이름이 생성됩니다. - 도구 설명: LLM을 위해 작업의
summary나description을 사용합니다. - API 세부 정보: 필요한 HTTP 메서드, 경로, 서버 기본 URL, 파라미터(경로, 쿼리, 헤더, 쿠키) 및 요청 본문 스키마를 내부적으로 저장합니다.
- 탐색된 각 작업에 대해
-
RestApiTool기능: 생성된 각RestApiTool은 다음을 수행합니다.- 스키마 생성: 작업의 파라미터와 요청 본문을 기반으로
FunctionDeclaration을 동적으로 생성합니다. 이 스키마는 LLM에게 도구를 호출하는 방법(예상되는 인수)을 알려줍니다. - 실행: LLM에 의해 호출되면, LLM이 제공한 인수와 OpenAPI 사양의 세부 정보를 사용하여 올바른 HTTP 요청(URL, 헤더, 쿼리 파라미터, 본문)을 구성합니다. 인증이 설정된 경우 이를 처리하고
httpx라이브러리를 사용하여 비동기적으로 API 호출을 실행합니다. - 응답 처리: API 응답(일반적으로 JSON)을 에이전트 플로우로 반환합니다.
- 스키마 생성: 작업의 파라미터와 요청 본문을 기반으로
-
인증:
OpenAPIToolset을 초기화할 때 전역 인증(예: API 키 또는 OAuth - 자세한 내용은 인증 참조)을 구성할 수 있습니다. 이 인증 구성은 생성된 모든RestApiTool인스턴스에 자동으로 적용됩니다.
사용 워크플로¶
OpenAPI 사양을 에이전트에 통합하려면 다음 단계를 따르세요.
- 사양 확보: OpenAPI 사양 문서(예:
.json또는.yaml파일에서 로드, URL에서 가져오기)를 준비합니다. -
툴셋 인스턴스화: 사양 내용과 타입(
spec_str/spec_dict,spec_str_type)을 전달하여OpenAPIToolset인스턴스를 생성합니다. 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) -
에이전트에 추가: 가져온 도구들을
LlmAgent의tools목록에 포함시킵니다. -
에이전트에게 지시: 에이전트의 지침을 업데이트하여 새로운 API 기능과 사용할 수 있는 도구 이름(예:
list_pets,create_pet)을 알려줍니다. 사양에서 생성된 도구 설명도 LLM에게 도움이 됩니다. - 에이전트 실행:
Runner를 사용하여 에이전트를 실행합니다. LLM이 API 중 하나를 호출해야 한다고 판단하면, 적절한RestApiTool을 대상으로 하는 함수 호출을 생성하고, 해당 도구가 자동으로 HTTP 요청을 처리합니다.
예제¶
이 예제는 간단한 펫 스토어 OpenAPI 사양(모의 응답을 위해 httpbin.org 사용)에서 도구를 생성하고 에이전트를 통해 상호작용하는 방법을 보여줍니다.
코드: 펫 스토어 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.")