REST API 외부 통합

GenD admin 이 발급한 M2M 서비스 클라이언트 (gend-svc-*) 자격증명으로 외부 서비스가 GenD REST API 를 호출하는 시나리오별 빠른 시작.

운영자 관점 매트릭스/스펙은 외부 연동 가이드 — base URL, 권한, 에러 shape 의 단일 진실 공급원. 본 튜토리얼은 외부 개발자가 자격증명만으로 시작하는 절차에 집중한다.

사전 조건

GenD admin 이 /admin/service-clients 에서 다음을 발급해 둔 상태.

항목	예시
Client ID	gend-svc-quant-ai
Client Secret	발급 모달에서 1회 표시되는 값
Role	viewer 또는 analyst
Base URL	https://gend.genon.ai

발급 모달을 닫으면 시크릿은 다시 볼 수 없다 — 즉시 시크릿 매니저에 저장.

공통 흐름

시나리오 A — curl

CI 잡, ad-hoc 디버깅, shell 스크립트.

환경변수

export GEND_BASE_URL="https://gend.genon.ai"
export GEND_CLIENT_ID="gend-svc-quant-ai"
export GEND_CLIENT_SECRET="<발급 모달에서 복사>"

토큰 발급

TOKEN=$(curl -sS -X POST \
  "$GEND_BASE_URL/auth/realms/gend/protocol/openid-connect/token" \
  -d grant_type=client_credentials \
  -d client_id="$GEND_CLIENT_ID" \
  -d client_secret="$GEND_CLIENT_SECRET" \
  | jq -r .access_token)

echo "${TOKEN:0:40}..."

API 호출

# 헬스체크 (인증 불필요)
curl -sS "$GEND_BASE_URL/health"

# 카탈로그 목록 (viewer)
curl -sS -H "Authorization: Bearer $TOKEN" \
  "$GEND_BASE_URL/api/v1/catalog/catalogs"

# SQL 실행 (analyst)
curl -sS -X POST -H "Authorization: Bearer $TOKEN" \
  -H "Content-Type: application/json" \
  -d '{"sql": "SELECT count(*) FROM hive.default.users"}' \
  "$GEND_BASE_URL/api/v1/query/execute"

# OpenAPI 스펙 (인증 불필요)
curl -sS "$GEND_BASE_URL/openapi.json" -o gend-openapi.json

GET /health 는 prefix 없음 — GET /api/v1/health 는 의도적으로 404. 자세한 매트릭스는 엔드포인트 매트릭스 참조.

시나리오 B — Python httpx

장기 실행 서비스, 데이터 파이프라인, FastAPI 백엔드의 GenD 어댑터.

설치

pip install httpx

클라이언트 클래스 (토큰 캐시 + 401 retry)

# gend_client.py
import os
import time
import httpx


class GendClient:
    """GenD REST API client with M2M token caching and 401 retry."""

    def __init__(self):
        self.base = os.environ["GEND_BASE_URL"].rstrip("/")
        self.realm = os.environ.get("GEND_REALM", "gend")
        self.client_id = os.environ["GEND_CLIENT_ID"]
        self.client_secret = os.environ["GEND_CLIENT_SECRET"]
        self._token: str | None = None
        self._exp: float = 0.0

    def _ensure_token(self) -> None:
        if self._token and time.time() < self._exp - 60:
            return
        r = httpx.post(
            f"{self.base}/auth/realms/{self.realm}/protocol/openid-connect/token",
            data={
                "grant_type": "client_credentials",
                "client_id": self.client_id,
                "client_secret": self.client_secret,
            },
            timeout=10,
        )
        r.raise_for_status()
        body = r.json()
        self._token = body["access_token"]
        self._exp = time.time() + body["expires_in"]

    def _request(self, method: str, path: str, **kw) -> httpx.Response:
        self._ensure_token()
        url = f"{self.base}{path}"
        headers = {"Authorization": f"Bearer {self._token}", **kw.pop("headers", {})}
        r = httpx.request(method, url, headers=headers, timeout=30, **kw)
        if r.status_code == 401:
            # 토큰 회전/만료 후 한 번 더 시도
            self._token = None
            self._exp = 0
            self._ensure_token()
            headers["Authorization"] = f"Bearer {self._token}"
            r = httpx.request(method, url, headers=headers, timeout=30, **kw)
        r.raise_for_status()
        return r

    def list_catalogs(self) -> list[dict]:
        return self._request("GET", "/api/v1/catalog/catalogs").json()

    def execute_query(self, sql: str) -> dict:
        return self._request(
            "POST", "/api/v1/query/execute", json={"sql": sql}
        ).json()

    def ask(self, question: str) -> dict:
        return self._request(
            "POST", "/api/v1/ai/ask", json={"question": question}
        ).json()

사용

from gend_client import GendClient

client = GendClient()
print(client.list_catalogs())
print(client.execute_query("SELECT 1 AS ping"))

토큰은 만료 60초 전까지 재사용되고, 401 응답 시 자동 재발급+재시도.

에러 응답 처리

GenD 는 현재 두 가지 shape 을 반환 — string 또는 Pydantic ValidationError list:

def parse_detail(response: httpx.Response) -> str:
    try:
        body = response.json()
        detail = body.get("detail", "")
        if isinstance(detail, list):
            # Pydantic ValidationError
            return "; ".join(d.get("msg", "") for d in detail)
        return str(detail)
    except (ValueError, AttributeError):
        return response.text[:200]

RFC 7807 Problem Details 표준화 후속 이슈에서 {type, title, status, detail, instance, code} 로 통일 예정. 그 전까지는 위 분기 패턴을 사용한다.

시나리오 C — LangChain Agent Tool

LLM 에이전트가 GenD 카탈로그·NL2SQL·RAG 를 도구로 호출하는 패턴.

설치

pip install langchain langchain-anthropic httpx

도구 정의

# gend_tools.py
from langchain.tools import tool
from gend_client import GendClient

_client = GendClient()


@tool
def list_gend_catalogs() -> list[dict]:
    """List all data catalogs available in GenD."""
    return _client.list_catalogs()


@tool
def gend_sql(sql: str) -> dict:
    """Execute a Trino SQL query against GenD. Read-only.

    Args:
        sql: Trino SQL — must be SELECT only, no DDL/DML.
    """
    return _client.execute_query(sql)


@tool
def gend_ask(question: str) -> dict:
    """Ask GenD in natural language — NL2SQL converts to Trino SQL and executes.

    Args:
        question: Natural language question about the data, e.g. "지난 주 주문 수".
    """
    return _client.ask(question)

Agent 구성

from langchain.agents import AgentExecutor, create_tool_calling_agent
from langchain_anthropic import ChatAnthropic
from langchain_core.prompts import ChatPromptTemplate

from gend_tools import gend_ask, gend_sql, list_gend_catalogs

llm = ChatAnthropic(model="claude-sonnet-4-6", temperature=0)

prompt = ChatPromptTemplate.from_messages([
    ("system", "You are a data analyst. Use GenD tools to answer questions about the data."),
    ("human", "{input}"),
    ("placeholder", "{agent_scratchpad}"),
])

tools = [list_gend_catalogs, gend_ask, gend_sql]
agent = create_tool_calling_agent(llm, tools, prompt)
executor = AgentExecutor(agent=agent, tools=tools, verbose=True)

result = executor.invoke({"input": "오늘 가장 많이 팔린 상품 카테고리 5개를 알려줘"})
print(result["output"])

GenD 의 NL2SQL (/api/v1/ai/ask) 가 자연어를 Trino SQL 로 변환하고 실행한다 — LLM 이 SQL 을 직접 작성할 필요 없음. SQL 실행 권한을 더 좁게 가져가려면 gend_ask 만 노출하고 gend_sql 은 제거.

권한 매트릭스 요약

호출	필요 role
GET /health, GET /openapi.json	(인증 불필요)
GET /api/v1/catalog/*, quality/*, governance/*, feature/*	viewer
POST /api/v1/query/execute, ai/ask, rag/query	analyst
POST/PUT/DELETE 변경 작업	admin (M2M client 자동 거부 — admin role 분리 발급 필요)

전체 매트릭스는 외부 연동 가이드의 권한 매트릭스.

토큰 라이프사이클

항목	값
Access Token 수명	30분 — 단, 클라이언트는 응답의 expires_in 을 사용할 것
Refresh Token	사용 안 함 (client_credentials 표준)
갱신 전략	expires_in 기준 만료 30~60초 전 재발급, 401 응답 시 한 번 더

트러블슈팅

증상	원인 / 해결
400 unauthorized_client	client_secret 불일치 — admin 의 secret 회전 후 외부 미반영
401 Token expired	토큰 만료 (30분) — 클라이언트 캐시 갱신
401 Invalid audience	service client 의 gend-api audience mapper 누락 (UI 발급 시 자동)
403 권한 부족	service-account realm role 미부여 — admin 에 부여 절차 요청
변경 작업 거부	M2M 토큰은 자동 admin 승격 없음 — admin role 분리 client 발급 필요
/openapi.json 이 HTML 반환	(구버전 ingress) — PR #662 머지 후 정상 JSON

다음 단계

• gend-cli 빠른 시작 — 같은 자격증명으로 터미널에서 호출
• MCP 클라이언트 연동 — 같은 자격증명으로 Claude Desktop / Cursor 에서 자연어 호출
• 외부 연동 가이드 — 단일 진실 공급원 (스펙/권한/에러 shape)
• M2M 서비스 계정 인증 — 발급/회전 UI 사용법