Python

서버 SDK백엔드에서 실행 (Calling Key)

Python SDK는 호출 시점의 사용자와 기능 컨텍스트를 게이트웨이 요청에 자동으로 실어 보냅니다. 토큰, 비용, 응답 속도와 요청 식별자는 게이트웨이가 기록합니다.

import os

from abto import init_abto

abto = init_abto(
    api_key=os.environ["ABTO_CALLING_KEY"],
    gateway_base_url="https://gateway.abto.app/v1",
    provider_keys={
        "openai": os.environ["OPENAI_API_KEY"],
        # 프로젝트가 여러 provider로 라우팅되면 후보 키를 함께 전달합니다.
        "anthropic": os.environ["ANTHROPIC_API_KEY"],
        "gemini": os.environ["GEMINI_API_KEY"],
        "deepseek": os.environ["DEEPSEEK_API_KEY"],
        "kimi": os.environ["KIMI_API_KEY"],
    },
)

openai = abto.openai()

with abto.with_context(
    device_id="device-abc",
    feature_id="review.summary",
):
    response = openai.chat.completions.with_raw_response.create(
        model="gpt-4.1-mini",
        messages=[{"role": "user", "content": "이 상품의 리뷰 12개를 세 줄로 요약해 줘."}],
    )
    request_id = response.headers.get("x-abto-request-id")
    completion = response.parse()

completion: 평소와 같은 OpenAI 응답 본문
raw response: Gateway가 발급한 x-abto-request-id를 읽는 자리
gateway_base_url: 필수. 생략하면 ABTO_GATEWAY_BASE_URL 환경 변수를 읽고, 둘 다 없으면 ValueError

LangChain에 연결하기

기존 ChatOpenAI에 ABTO의 연결 설정을 전달하면 됩니다. 체인, 검색기, 프롬프트, 콜백과 재시도 설정은 그대로 사용하세요.

pip install "abto[openai]" langchain-openai

아래는 동기 호출과 비동기 호출을 모두 연결하는 예제입니다. Calling Key와 OpenAI API 키는 서버 환경변수에 설정하세요.

import asyncio
import os

from abto import init_abto
from langchain_openai import ChatOpenAI

abto = init_abto(
    api_key=os.environ["ABTO_CALLING_KEY"],
    gateway_base_url="https://gateway.abto.app/v1",
    provider_keys={"openai": os.environ["OPENAI_API_KEY"]},
)

sync_options = abto.openai_options()
async_options = abto.async_openai_options()
model = ChatOpenAI(
    model="gpt-4.1-mini",
    use_responses_api=False,
    **sync_options,
    http_async_client=async_options["http_client"],
)

async def main():
    try:
        with abto.with_context(feature_id="support.reply"):
            reply = await model.ainvoke("배송 상태를 확인하는 방법을 알려 줘.")
            print(reply.content)
    finally:
        sync_options["http_client"].close()
        await async_options["http_client"].aclose()

asyncio.run(main())

동기 코드에서는 같은 모델의 model.invoke(...)를 사용하세요. 서버에서는 모델과 연결 설정을 한 번 만들고 재사용한 뒤, 서버 종료 시 HTTP 클라이언트를 닫으세요. 호출 후 대시보드의 요청 목록에서 support.reply를 찾아 응답, 비용, 응답 속도를 확인할 수 있습니다.

기존 HTTP 클라이언트가 있다면

기존 클라이언트를 만드는 곳에서 transport를 wrap_httpx_transport()로 감싼 뒤 ABTO에 전달하세요. 기존 send() 재정의, 요청과 응답 훅, 연결 풀과 쿠키를 계속 사용합니다.

import httpx
from abto import wrap_httpx_transport

# 기존 transport가 있다면 HTTPTransport() 대신 그 객체를 감싸세요.
existing_sync_client = httpx.Client(
    transport=wrap_httpx_transport(httpx.HTTPTransport()),
    trust_env=False,
)
sync_options = abto.openai_options(http_client=existing_sync_client)

existing_async_client = httpx.AsyncClient(
    transport=wrap_httpx_transport(httpx.AsyncHTTPTransport()),
    trust_env=False,
)
async_options = abto.async_openai_options(http_client=existing_async_client)

기존 timeout, TLS, 프록시 설정을 유지하세요. 예시는 환경변수의 자동 프록시 선택을 끕니다. 프록시가 필요하면 기존 프록시를 transport에 명시하세요. mounts를 사용한다면 해당 transport도 각각 감싸야 합니다. 이미 생성한 클라이언트의 내부 설정을 ABTO가 교체하지 않으므로, 클라이언트를 생성하는 코드에서 연결해야 합니다. 설정이 빠진 전송 경로에는 ABTO 키를 넣지 않으며, 응답을 받은 뒤 설정 오류를 알립니다.

ABTO 설정을 닫아도 전달한 클라이언트는 닫히지 않습니다. 기존 클라이언트는 원래 종료 처리에서 닫으세요. Gateway와 직접 폴백의 인증은 ABTO가 설정하며, 키가 다른 목적지로 전달되지 않도록 자동 리다이렉트를 끕니다. 고객 훅이 실행된 뒤 목적지와 Host 헤더를 검사하고, 전송용 요청에만 키를 넣습니다. 기존 요청 훅, 응답 훅과 send()에는 실제 ABTO 키를 추가하지 않습니다. 고객 훅이나 send() 안의 관측 코드에서 발생한 오류는 ABTO의 직접 폴백을 일으키지 않고 그대로 전달됩니다. OpenAI와 LangChain의 기존 재시도 설정은 그대로 적용됩니다. 직접 구성한 로깅 transport는 ABTO wrapper 바깥에 두세요. 실제 네트워크 전송을 맡은 안쪽 transport에는 인증 키가 전달됩니다.

openai_options()는 공식 OpenAI에도, async_openai_options()는 공식 AsyncOpenAI에도 전달할 수 있습니다. 반환된 api_key는 추적 데이터에 실제 키가 남지 않도록 사용하는 placeholder입니다. 실제 키는 요청을 보낼 때 주입합니다.

요청별 컨텍스트와 추적 유지하기

사용자마다 with abto.with_context(feature_id=..., device_id=...) 안에서 체인을 호출하세요. 비동기 동시 요청에서도 contextvars로 컨텍스트를 구분합니다. 브라우저에서 받은 device ID는 백그라운드 작업의 데이터에도 함께 저장해 전달해야 합니다.

Langfuse와 LangSmith 콜백, tags와 metadata는 기존 호출 설정에 그대로 두세요. 추적 도구에는 앱이 보낸 입력이 기록됩니다. 대시보드에서 바꾼 모델, 프롬프트, 파라미터의 실제 실행 내용은 ABTO 요청 기록에서 확인하세요. Gateway가 제공사로 전달할 때 추적 헤더를 제거할 수 있으므로, 제공사까지 하나의 trace로 연결되는 것은 보장하지 않습니다.

이 연결은 OpenAI Chat Completions용입니다. Streaming과 Responses API를 사용하는 호출은 이 예제로 전환하지 마세요. TXT 로더와 벡터 검색기는 기존 체인에 그대로 두고, 임베딩은 기존 제공사 연결을 유지하세요. Gateway 장애 시 직접 호출을 이어가려면 아래의 direct fallback 설정을 함께 사용하세요.

요청마다 전송되는 Gateway 헤더

SDK 입력	Gateway 헤더	의미
`api_key`	`Authorization: Bearer …`	ABTO Calling Key
`provider_keys["openai"]`	`x-abto-key-openai`	직접 계약한 OpenAI provider key
`provider_keys["gemini"]`	`x-abto-key-gemini`	직접 계약한 Gemini provider key
`provider_keys["anthropic"]`	`x-abto-key-anthropic`	직접 계약한 Anthropic provider key
`provider_keys["deepseek"]`	`x-abto-key-deepseek`	직접 계약한 DeepSeek provider key
`provider_keys["kimi"]`	`x-abto-key-kimi`	직접 계약한 Kimi provider key
`feature_id`	`x-abto-feature-id`	기능 ID
`node_id`	`x-abto-node-id`	기능 안의 노드 (선택)
`device_id`	`x-abto-device-id`	사용자 기기 식별자 (선택)
`trace_id`	`x-abto-trace-id`	실행 ID (넣지 않으면 자동)

provider_keys의 성격과 취급:

Gateway에 키를 등록하는 설정이 아니라, 서버에 보관한 credential을 요청마다 헤더로 실어 보내는 입력
요청 범위로만 전달. 호출 기록에는 미저장
라우팅된 provider의 키가 없으면 Gateway가 그 요청을 거절
문자열 대신 함수를 주면 요청마다 다시 평가. key rotation과 provider별 credential resolver에 사용

호출자가 넣은 헤더의 처리:

abto.openai()가 만든 client에서는 extra_headers로 x-abto-key-*를 덮어쓰지 마세요.
Python SDK가 호출자의 Authorization, x-abto-key-*, x-abto-feature-id, x-abto-node-id, x-abto-device-id, x-abto-trace-id를 제거하고 init_abto의 신뢰된 설정과 context로 다시 구성합니다.
Python SDK 없이 기존 OpenAI SDK를 직접 쓴다면 같은 헤더를 각 요청의 extra_headers에 직접 넣습니다. 직접 연결 예시를 참고하세요.

컨텍스트 필드

필드의 의미는 Node.js JavaScript와 같습니다.

feature_id: 기능 ID(예: review.summary). x-abto-feature-id로 전송
node_id: 기능 안에서 LLM을 부르는 지점(예: review.draft). 없으면 기능 ID가 노드 이름이 됨. x-abto-node-id로 전송
- 예: with abto.with_context(feature_id="review.summary", node_id="review.draft"):
- 한 기능이 LLM을 여러 번 부르면 호출 지점마다 다른 값을 넣어, 라우팅에서 노드마다 에이전트 설정을 따로 고를 수 있음
- 안쪽 with_context에서 바깥과 다른 feature_id를 넣고 node_id를 넣지 않으면, 바깥의 node_id를 물려받지 않음
device_id: x-abto-device-id로 전송. 제품 행동과 잇기 위해 브라우저가 만든 device_id를 백엔드로 받아 그대로 전달
trace_id: 에이전트나 워크플로 실행 한 번이 만든 호출을 하나로 묶는 실행 ID. x-abto-trace-id로 전송
- 넣지 않으면 가장 바깥 with_context가 새 실행 ID를 만들고, 안쪽 with_context는 그 값을 물려받음
- 실행 하나를 통째로 with_context 하나로 감싸야 그 안의 모든 호출이 같은 실행으로 묶임. 호출마다 따로 감싸면 호출마다 실행이 나뉨
- with_context 밖에서 나간 호출은 호출마다 새 실행 ID를 받음
- get_headers(AbtoContext(...))처럼 context를 직접 넘길 때 trace_id를 비우면 감싼 with_context의 값을 씀
- 다른 서비스가 넘겨준 실행을 이어 갈 때만 가장 바깥 with_context에 그 값을 직접 넣음
- 200자 이하의 출력 가능한 ASCII(영문, 숫자, 공백, 기호)만 받음. 비어 있거나 규칙을 벗어난 값은 Gateway가 오류 없이 무시함
- 요청의 traceparent는 실행 ID로 쓰지 않음

식별자를 다루는 규칙:

클라이언트가 보낸 device_id, feature_id는 애플리케이션의 기존 요청 스키마로 검증
Calling Key와 provider key는 클라이언트 요청에서 받지 않고 서버 환경 변수나 서버 전용 credential resolver에서만 읽기

응답을 브라우저와 모바일 행동과 잇는 방법:

위 예제처럼 with_raw_response로 x-abto-request-id를 읽기
백엔드 응답에 같은 헤더를 포함해 Browser 또는 Mobile SDK의 LLM trace에 연결

기존 추적 정보 유지

기존 추적 도구가 보낸 traceparent와 tracestate는 ABTO 호출과 직접 폴백에서도 유지됩니다. ABTO는 고객의 추적 ID를 덮어쓰지 않고 기능과 기기 정보를 별도로 추가합니다.

이 동작은 1.1.2부터 지원합니다. 추적 도구의 요청 헤더 설정은 유지하세요. ABTO가 추적 도구를 자동으로 설정하거나 제공사까지 추적을 연결하지는 않습니다.

실행 ID인 trace_id는 traceparent로 바꾸지 않고 x-abto-trace-id로만 보냅니다.

Gateway가 지원하는 OpenAI 요청

데이터 경로는 현재 OpenAI Chat Completions
user 메시지의 인라인 base64 이미지와 PDF 지원
Function tool calling 지원. tool 결과를 되돌려 보낼 때 tool_call_id는 응답의 message.tool_calls[].id를 그대로 사용(최상위 id가 아님)
Streaming, 원격 이미지 URL, audio 등 미지원 필드는 조용히 무시하지 않고 400으로 거절
정확한 필드 목록은 Gateway Chat Completions 참고

재시도

openai = abto.openai(max_retries=2)

공식 OpenAI SDK의 max_retries를 그대로 씁니다. ABTO는 이 값을 덮어쓰지 않습니다
설정하지 않으면 공식 OpenAI SDK의 기본값을 그대로 사용
Calling SDK는 자체 재시도 횟수를 두지 않습니다
direct fallback에는 횟수 대신 재전송 여부 스위치 fallback.on_timeout 하나만 존재. 기본값은 꺼짐
그 밖의 공식 OpenAI option도 그대로 전달
abto.openai()에서는 api_key, base_url, http_client를 신뢰된 라우팅을 위해 ABTO가 소유. 넘기면 조용히 무시하지 않고 ValueError 발생

Gateway 장애 시 OpenAI direct fallback

ABTO 도입 이전에 쓰던 엔드포인트로 되돌아가는 비상 경로입니다. 그래서 목적지를 fallback.base_url에 직접 지정해야 하며, 기본값은 없습니다. OpenAI에서 직접 키를 발급받아 공식 SDK를 쓰던 고객이라면 그 주소가 https://api.openai.com/v1입니다.

from abto import OpenAIDirectFallbackOptions, init_abto

abto = init_abto(
    api_key=os.environ["ABTO_CALLING_KEY"],
    gateway_base_url="https://gateway.abto.app/v1",
    provider_keys={"openai": os.environ["OPENAI_API_KEY"]},
    fallback=OpenAIDirectFallbackOptions(
        # ABTO를 붙이기 전 이 코드가 호출하던 주소를 그대로 적습니다.
        base_url="https://api.openai.com/v1",
        timeout_seconds=30,
        on_timeout=False,
    ),
)

원본 Chat Completions body와 model을 그대로 그 주소로 보내며, Gateway의 provider/model 정책을 재현하지 않습니다.

현재 요청을 direct로 보내는 경우

ABTO에 요청이 닿지 못한 경우
닿았어도 provider를 부르기 전에 거절된 경우

즉 provider가 실행되지 않은 것이 확실할 때만 전환합니다. 이중 실행과 이중 과금을 만들지 않습니다.

현재 요청을 폴백하지 않는 경우

timeout과 전송 여부가 모호한 disconnect. Gateway가 이미 provider를 실행했을 가능성
provider와 transport 오류, 결정적인 4xx와 429, streaming 시작 이후
이때 direct circuit도 열지 않으며, 공식 OpenAI SDK가 재시도하면 다시 Gateway를 호출

설정

base_url: 필수. ABTO 이전에 쓰던 주소. OpenAI 요청 경로와 Authorization: Bearer를 받는 엔드포인트여야 합니다
- 이 값을 적는 것이 폴백을 켜는 방법입니다. 별도의 활성 플래그는 없습니다
- 목적지 없이 fallback=True만 주거나, OpenAI key 없이 목적지만 주면 init_abto가 ValueError를 발생시킵니다
timeout_seconds: Gateway connection pool 대기, 연결, 쓰기, 응답 헤더 읽기 각 단계의 inactivity 상한. 헤더 이후 body와 direct 요청은 abto.openai(timeout=...) 값을 유지
on_timeout=True: timeout 난 현재 요청까지 재전송. 중복 실행과 과금 위험을 감수하는 명시적 선택
fallback=False: 기능 전체 해제

경계

transport는 SDK attempt마다 Gateway 판단과 direct 전송을 각각 한 번만 수행
direct 호출은 Gateway policy, ABTO telemetry, request_id를 거치지 않음. model 은 그 엔드포인트가 지원하는 것으로 제한
OpenAI에 필요한 header만 전달하고 Cookie, proxy credential, 사용자 정의 Gateway header는 제거
direct 전환 뒤의 응답과 오류는 공식 OpenAI SDK로 그대로 반환하며 재시도 여부는 그 SDK 설정이 결정
Anthropic과 Gemini의 네이티브 direct fallback은 현재 범위 밖