openai python package에서 내가 보내는 패킷을 가능한 raw 데이터 형태로 보고 싶어

좋아요. “openai” Python 패키지로 API를 호출할 때, 내가 보낸 HTTP 요청(=패킷)에 최대한 가까운 raw 데이터 를 보는 대표적인 방법을 단계별로 정리했어요. 상황/보안 요구에 따라 골라 쓰시면 됩니다.

1) 가장 간단한 방법: SDK 디버그 로그 켜기

OpenAI Python SDK는 환경변수 한 줄로 상세 로그를 뿌릴 수 있어요.

export OPENAI_LOG=debug
python your_script.py
  • 이러면 SDK 내부와 함께 httpx(OpenAI SDK의 HTTP 클라이언트) 디버그 로그도 활성화돼서 요청/응답 흐름을 확인할 수 있습니다. Stack Overflow

장점: 코드 수정 없음
단점: 바디 전문을 항상 보여주진 않음(민감정보 보호 차원에서 마스킹/생략될 수 있음)

2) 표준 로깅으로 httpx 디버깅(요청/응답 라인·헤더)

logging으로 httpx 로거를 DEBUG로 켜면, 연결/요청/응답의 내부 동작을 자세히 볼 수 있어요.

import logging

logging.basicConfig(level=logging.DEBUG)
logging.getLogger("httpx").setLevel(logging.DEBUG)

from openai import OpenAI
client = OpenAI()

# 평소처럼 호출
resp = client.chat.completions.create(
    model="gpt-4o-mini",
    messages=[{"role": "user", "content": "ping"}],
)
print(resp)
  • 위처럼 하면 httpx가 출력하는 요청 메서드/URL, 응답 상태, 일부 헤더 등 을 확인할 수 있습니다. 필요한 경우 포맷을 더 꾸밀 수 있어요. python-httpx.org

장점: 표준 파이썬 로깅으로 손쉽게 제어
단점: 바디 전문은 기본 로그에 안 나옴

3) “진짜 raw에 가깝게” — 커스텀 httpx Client + 이벤트 훅

가장 투명한 방법은 SDK에 커스텀 httpx.Client 를 주입하고, event_hooks 로 요청·응답 시점에 바디까지 포함 해서 찍는 겁니다.

import json
import httpx
from openai import OpenAI

def log_request(request: httpx.Request):
    # 주의: 여기서 body를 출력하면 프롬프트/키 등 민감정보가 노출될 수 있음
    print("\n=== HTTP REQUEST ===")
    print(f"{request.method} {request.url}")
    for k, v in request.headers.items():
        print(f"{k}: {v}")
    # request.content는 bytes일 수 있음
    if request.content:
        try:
            print("\n-- body (utf-8) --")
            print(request.content.decode("utf-8"))
        except UnicodeDecodeError:
            print("\n-- body (bytes) --")
            print(request.content)

def log_response(response: httpx.Response):
    print("\n=== HTTP RESPONSE ===")
    print(f"HTTP/{response.http_version} {response.status_code}")
    for k, v in response.headers.items():
        print(f"{k}: {v}")
    # 주의: 응답 바디에는 토큰/텍스트 등 민감정보 포함 가능
    body = response.content
    try:
        print("\n-- body (utf-8 pretty) --")
        print(json.dumps(response.json(), ensure_ascii=False, indent=2))
    except Exception:
        print("\n-- body (raw) --")
        print(body[:2000])  # 너무 길면 샘플만

hooks = {
    "request": [log_request],
    "response": [log_response],
}
http_client = httpx.Client(event_hooks=hooks)

client = OpenAI(http_client=http_client)

# 테스트 호출
_ = client.chat.completions.create(
    model="gpt-4o-mini",
    messages=[{"role": "user", "content": "Hello"}],
)
  • 핵심은 OpenAI(http_client=...) 로 내 httpx 클라이언트를 주입 하는 부분입니다. 이런 식으로 요청/응답 헤더 + 바디 를 원하는 형식으로 덤프할 수 있어요. TIL+2

장점: 가장 “raw”에 가까움(내용·헤더·상태 전부 내가 통제)
단점: 직접 마스킹 처리를 해야 안전

4) 외부 프록시(패킷 캡처)로 “와이어 그대로” 보기

코드를 건드리고 싶지 않거나, 진짜 전송되는 TLS 트래픽 까지 관찰하고 싶다면 프록시/패킷 도구 를 쓰세요.

  • 예: Proxyman , mitmproxy , Charles , Fiddler 등

  • 방법: 도구에서 루트 인증서를 신뢰하도록 설정 → Python이 해당 프록시를 쓰게 HTTPS_PROXY/HTTP_PROXY 환경변수 설정 → 스크립트 실행 → 도구에서 요청 라인/헤더/압축 전/후 바디 까지 열람

  • httpx/requests 기반 파이썬 코드에서 이런 프록시로 HTTPS 트래픽 캡처 가 가능합니다. Proxyman

장점: SDK/코드 수정 최소화, 실제 전송 형태를 GUI로 직관적으로 확인
단점: 사내 정책/루트 인증서 설치 이슈 가능

5) 비교 기준 정리 (언제 무엇을 쓸까?)

  • 빠른 확인 : OPENAI_LOG=debug (1분 컷). Stack Overflow

  • 파이썬 로그에 일원화 : httpx 로거 DEBUG (Step 2). python-httpx.org

  • 요청/응답 바디까지 100% 통제 : 커스텀 httpx Client + event_hooks (Step 3). TIL+1

  • 진짜 패킷 관점(프록시) : Proxyman/mitmproxy (Step 4). Proxyman

6) 보안/컴플라이언스 주의(중요)

  • API 키 와 프롬프트/응답 내용 이 로그에 남을 수 있어요.

    • 로컬만, 접근 제한된 환경에서만 출력하세요.

    • 저장 시엔 마스킹/암호화 또는 임시 파일 사용을 권장.

  • 회사/조직 정책상 중간자 프록시 툴 사용이 제한될 수 있습니다(특히 클라우드/생산환경). 사전 승인 필요.

7) 참고 문서

원하시면, 위 ③ 커스텀 클라이언트 코드를 바디 마스킹(예: 토큰/이메일 제거) 포함 버전으로 바로 만들어 드릴게요.

string에 데이터가 담기게 하고 싶어 요청을 보낼때 / 받을때 각각

좋습니다 — 요청 보낼 때(request)와 받을 때(response) 각각의 바디를 문자열(string)으로 저장하는 실전 코드(동기 + 비동기)를 바로 드릴게요. 민감정보(Authorization/API Key 등) 마스킹도 포함되어 있으니 안전하게 쓰실 수 있습니다. 단계별로 설명도 덧붙였어요 — 요청하신 대로 step-by-step 방식으로요.

1) 핵심 아이디어 (간단 요약)

  1. httpx.Client(또는 httpx.AsyncClient)를 만들어 event_hooks에 요청/응답 훅을 붙입니다.

  2. 훅에서 request.content / response.content를 읽어 str로 변환해서 전역 변수나 캡처용 변수에 저장합니다.

  3. 로그로 출력하기 전 민감 필드(헤더의 Authorization 등)와 바디 내부 키(예: "api_key", "key")를 마스킹합니다.

2) 동기(synchronous) 예제 — 바로 복사해서 쓰세요

import json
import httpx
from openai import OpenAI    # 설치된 최신 openai 패키지 사용

# 저장할 문자열 변수(요청/응답)
captured_request_body = None
captured_response_body = None

def redact_headers(headers: dict) -> dict:
    redacted = {}
    for k, v in headers.items():
        if k.lower() == "authorization" or "api-key" in k.lower():
            redacted[k] = "REDACTED"
        else:
            redacted[k] = v
    return redacted

def redact_json_text(text: str) -> str:
    # 시도: JSON이면 파싱해서 특정 키 마스킹
    try:
        obj = json.loads(text)
        if isinstance(obj, dict):
            for sensitive in ("api_key", "apiKey", "key", "authorization", "password"):
                if sensitive in obj:
                    obj[sensitive] = "REDACTED"
        return json.dumps(obj, ensure_ascii=False)
    except Exception:
        # JSON이 아니면 간단 치환(헤더처럼)
        # 예: URL-encoded form/body에서 key=값 형태를 가볍게 마스킹
        for sensitive in ("api_key", "apiKey", "key", "authorization", "password"):
            text = text.replace(sensitive + "=", sensitive + "=REDACTED")
        return text

def request_hook(request: httpx.Request):
    global captured_request_body
    # 헤더 마스킹
    safe_headers = redact_headers(dict(request.headers))
    # 바디 읽기 (bytes -> str)
    body_str = None
    # request.content는 bytes 또는 None
    if request.content is not None:
        try:
            body_str = request.content.decode("utf-8", errors="replace")
        except Exception:
            body_str = repr(request.content)
    else:
        # 준비되지 않은 스트림(rare)인 경우
        body_str = "<no request.content available or streaming body>"
    # 민감정보 마스킹
    body_str = redact_json_text(body_str)
    captured_request_body = body_str

    # (선택) 콘솔 출력
    print("---- REQUEST ----")
    print(f"{request.method} {request.url}")
    for k, v in safe_headers.items():
        print(f"{k}: {v}")
    print("-- body --")
    print(body_str)
    print("-----------------")

def response_hook(response: httpx.Response):
    global captured_response_body
    # 헤더 마스킹
    safe_headers = redact_headers(dict(response.headers))
    # 응답 바디 읽음 (bytes -> str)
    try:
        # response.content는 전체 바디(이미 수신된 상태의 hook에서는 안전)
        body_str = response.content.decode("utf-8", errors="replace")
        # 시도해서 JSON 예쁘게
        try:
            parsed = json.loads(body_str)
            body_str = json.dumps(parsed, ensure_ascii=False, indent=2)
        except Exception:
            pass
    except Exception:
        body_str = repr(response.content)
    body_str = redact_json_text(body_str)
    captured_response_body = body_str

    print("---- RESPONSE ----")
    print(f"HTTP/{response.http_version} {response.status_code}")
    for k, v in safe_headers.items():
        print(f"{k}: {v}")
    print("-- body --")
    print(body_str[:4000])  # 너무 길면 잘라서 보이게
    print("------------------")

# 이벤트 훅 등록
hooks = {
    "request": [request_hook],
    "response": [response_hook],
}
http_client = httpx.Client(event_hooks=hooks, timeout=60.0)

# OpenAI 클라이언트에 주입
client = OpenAI(http_client=http_client)

# 실제 요청 (예시)
resp = client.chat.completions.create(
    model="gpt-4o-mini",
    messages=[{"role": "user", "content": "안녕"}],
)

# 함수 외부에서 문자열 접근 가능
print("\nCaptured request body (string):")
print(captured_request_body)
print("\nCaptured response body (string):")
print(captured_response_body)

3) 비동기(async) 예제 — async/await 환경에서

import json
import httpx
import asyncio
from openai import OpenAI

captured_request_body = None
captured_response_body = None

def redact_headers(headers: dict) -> dict:
    redacted = {}
    for k, v in headers.items():
        if k.lower() == "authorization" or "api-key" in k.lower():
            redacted[k] = "REDACTED"
        else:
            redacted[k] = v
    return redacted

def redact_json_text(text: str) -> str:
    try:
        obj = json.loads(text)
        if isinstance(obj, dict):
            for sensitive in ("api_key", "apiKey", "key", "authorization", "password"):
                if sensitive in obj:
                    obj[sensitive] = "REDACTED"
        return json.dumps(obj, ensure_ascii=False)
    except Exception:
        for sensitive in ("api_key", "apiKey", "key", "authorization", "password"):
            text = text.replace(sensitive + "=", sensitive + "=REDACTED")
        return text

async def request_hook(request: httpx.Request):
    global captured_request_body
    safe_headers = redact_headers(dict(request.headers))
    body_str = None
    if request.content is not None:
        try:
            body_str = request.content.decode("utf-8", errors="replace")
        except Exception:
            body_str = repr(request.content)
    else:
        body_str = "<no request.content available or streaming body>"
    body_str = redact_json_text(body_str)
    captured_request_body = body_str
    print("REQ", request.method, request.url)
    print(body_str)

async def response_hook(response: httpx.Response):
    global captured_response_body
    safe_headers = redact_headers(dict(response.headers))
    try:
        body_str = response.content.decode("utf-8", errors="replace")
    except Exception:
        body_str = repr(response.content)
    body_str = redact_json_text(body_str)
    captured_response_body = body_str
    print("RES", response.status_code)
    print(body_str[:2000])

async def main():
    hooks = {"request": [request_hook], "response": [response_hook]}
    async with httpx.AsyncClient(event_hooks=hooks, timeout=60.0) as async_http:
        client = OpenAI(http_client=async_http)
        resp = await client.chat.completions.create(
            model="gpt-4o-mini",
            messages=[{"role": "user", "content": "안녕"}],
        )
        print("Done")

asyncio.run(main())

4) 주의사항 & 팁 (중요)

  • 대부분 경우 request.content / response.content는 훅에서 읽을 수 있고 문자열로 변환 가능합니다. 다만 파일 업로드(멀티파트) 혹은 스트리밍 바디 라면 처리 방식이 달라집니다.

  • 스트리밍(예: chunked transfer / SSE)으로 동작하는 경우: 훅에서 전체 바디가 아직 완성되지 않았을 수 있습니다. 이런 상황이면 SDK의 스트리밍 API(또는 httpx의 stream() 사용)를 직접 감싸서 각 청크를 캡처하는 로직을 추가해야 합니다. 원하시면 스트리밍 캡처 예제도 드릴게요.

  • 민감정보 보호 : 콘솔/파일에 직접 남기면 안 됩니다. 로컬 전용, 접근 제어된 환경에서만 보관하세요. 저장해야 한다면 마스킹 또는 암호화 권장.

  • event_hooks에서 바디를 읽는 건 일반적으로 안전하지만, 일부 경우(특히 request가 generator/stream로 만들어졌다면) 읽으면 원래 전송을 망칠 수 있으니 주의하세요. 위 예제는 일반 JSON/폼/멀티파트 전송에 적합합니다.

5) 추가 요청 가능한 것들 (원하시면 바로 만듭니다)

  • 요청/응답을 파일 대신 문자열 변수 에 안전하게 누적(rolling buffer)하는 버전

  • 스트리밍(서버-전송 이벤트) 응답을 청크 단위로 캡처해서 단일 문자열로 합치는 예제

  • 바디에서 특정 키만(예: messages or prompt)만 캡처해서 저장하고 나머지는 마스킹하는 예제

  • 멀티파트(파일 업로드) 요청의 각 파트(파일 메타/콘텐츠)를 캡처해 문자열로 변환하는 예제

원하시는 걸 알려주시면(예: 스트리밍 캡처, 멀티파트 처리, 혹은 바로 적용 가능한 마스킹 규칙) 지금 바로 코드로 만들어 드릴게요. 어떤 걸 먼저 드릴까요?

전역 변수 말고 main 안에서 스트링 저장하는 방밥

좋아요 — 전역 변수 없이main() 안에서 요청/응답 바디를 문자열로 저장하는 방법을 바로 드릴게요. 핵심은 클로저(or 로컬 가변 컨테이너) 를 사용해 훅이 main 안의 변수에 접근하게 하는 것입니다. 스트리밍 예외는 마지막에 짧게 언급합니다.

아래 코드는 (1) 동기(synchronous)와 (2) 비동기(async) 두 버전 모두 제공하고, 민감정보(Authorization 헤더, 몇몇 바디 키)를 마스킹 합니다. 그대로 복사해서 쓰시면 됩니다.

1) 원리(요약, step-by-step)

  1. main() 안에 captured = {"request": None, "response": None} 같은 가변 dict 를 만든다.

  2. request_hook / response_hook 함수는 클로저로서 그 captured를 수정한다. (nonlocal 불필요 — dict는 가변이므로)

  3. 훅에서 request.content / response.content를 문자열로 변환 → 민감정보 마스킹 → captured[...] = body_str 저장.

  4. 요청이 끝난 뒤 main()에서 captured 값을 사용(또는 반환)한다.

2) 동기(sync) 예제

import json
import httpx
from openai import OpenAI

def redact_headers(headers: dict) -> dict:
    out = {}
    for k, v in headers.items():
        if k.lower() == "authorization" or "api-key" in k.lower():
            out[k] = "REDACTED"
        else:
            out[k] = v
    return out

def redact_json_text(text: str) -> str:
    try:
        obj = json.loads(text)
        if isinstance(obj, dict):
            for sensitive in ("api_key", "apiKey", "key", "authorization", "password"):
                if sensitive in obj:
                    obj[sensitive] = "REDACTED"
        return json.dumps(obj, ensure_ascii=False)
    except Exception:
        for sensitive in ("api_key", "apiKey", "key", "authorization", "password"):
            text = text.replace(sensitive + "=", sensitive + "=REDACTED")
        return text

def main():
    # 1) 로컬 캡처 컨테이너 (전역 아님)
    captured = {"request": None, "response": None}

    # 2) 훅들 (클로저로 captured에 접근)
    def request_hook(request: httpx.Request):
        # headers 마스킹(여기서는 저장용; 실제 출력은 안전하게)
        safe_headers = redact_headers(dict(request.headers))
        # body를 문자열로 읽기
        if request.content is not None:
            try:
                body_str = request.content.decode("utf-8", errors="replace")
            except Exception:
                body_str = repr(request.content)
        else:
            body_str = "<no request.content available or streaming body>"
        body_str = redact_json_text(body_str)
        captured["request"] = {
            "method": request.method,
            "url": str(request.url),
            "headers": safe_headers,
            "body": body_str,
        }

    def response_hook(response: httpx.Response):
        safe_headers = redact_headers(dict(response.headers))
        try:
            body_str = response.content.decode("utf-8", errors="replace")
            # 예쁘게 포맷(JSON이면)
            try:
                parsed = json.loads(body_str)
                body_str = json.dumps(parsed, ensure_ascii=False, indent=2)
            except Exception:
                pass
        except Exception:
            body_str = repr(response.content)
        body_str = redact_json_text(body_str)
        captured["response"] = {
            "status_code": response.status_code,
            "headers": safe_headers,
            "body": body_str,
        }

    hooks = {"request": [request_hook], "response": [response_hook]}
    http_client = httpx.Client(event_hooks=hooks, timeout=60.0)
    client = OpenAI(http_client=http_client)

    # 3) 실제 호출 (동기)
    resp = client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[{"role": "user", "content": "안녕하세요"}],
    )

    # 4) main 내부에서 문자열로 활용 / 출력 / 반환 가능
    print("Captured request body (string):")
    print(captured["request"]["body"] if captured["request"] else "<none>")
    print("\nCaptured response body (string):")
    print(captured["response"]["body"] if captured["response"] else "<none>")

    return captured

if __name__ == "__main__":
    captured = main()

3) 비동기(async) 예제

import json
import httpx
import asyncio
from openai import OpenAI

def redact_headers(headers: dict) -> dict:
    out = {}
    for k, v in headers.items():
        if k.lower() == "authorization" or "api-key" in k.lower():
            out[k] = "REDACTED"
        else:
            out[k] = v
    return out

def redact_json_text(text: str) -> str:
    try:
        obj = json.loads(text)
        if isinstance(obj, dict):
            for sensitive in ("api_key", "apiKey", "key", "authorization", "password"):
                if sensitive in obj:
                    obj[sensitive] = "REDACTED"
        return json.dumps(obj, ensure_ascii=False)
    except Exception:
        for sensitive in ("api_key", "apiKey", "key", "authorization", "password"):
            text = text.replace(sensitive + "=", sensitive + "=REDACTED")
        return text

async def main():
    captured = {"request": None, "response": None}

    async def request_hook(request: httpx.Request):
        safe_headers = redact_headers(dict(request.headers))
        if request.content is not None:
            try:
                body_str = request.content.decode("utf-8", errors="replace")
            except Exception:
                body_str = repr(request.content)
        else:
            body_str = "<no request.content available or streaming body>"
        body_str = redact_json_text(body_str)
        captured["request"] = {
            "method": request.method,
            "url": str(request.url),
            "headers": safe_headers,
            "body": body_str,
        }

    async def response_hook(response: httpx.Response):
        safe_headers = redact_headers(dict(response.headers))
        try:
            body_str = response.content.decode("utf-8", errors="replace")
            try:
                parsed = json.loads(body_str)
                body_str = json.dumps(parsed, ensure_ascii=False, indent=2)
            except Exception:
                pass
        except Exception:
            body_str = repr(response.content)
        body_str = redact_json_text(body_str)
        captured["response"] = {
            "status_code": response.status_code,
            "headers": safe_headers,
            "body": body_str,
        }

    hooks = {"request": [request_hook], "response": [response_hook]}

    async with httpx.AsyncClient(event_hooks=hooks, timeout=60.0) as async_http:
        client = OpenAI(http_client=async_http)
        await client.chat.completions.create(
            model="gpt-4o-mini",
            messages=[{"role": "user", "content": "안녕하세요 (async)"}],
        )

    print("Captured request (local):", captured["request"]["body"] if captured["request"] else "<none>")
    print("Captured response (local):", captured["response"]["body"] if captured["response"] else "<none>")
    return captured

if __name__ == "__main__":
    asyncio.run(main())

4) 주의사항(중요)

  • 스트리밍 응답(SSE/청크) 또는 파일 업로드(멀티파트) 의 경우 request.content / response.content가 훅에서 바로 전체로 읽히지 않거나, 읽으면 스트림이 소모되어 전송에 영향이 생길 수 있음. 그런 경우는 별도 스트리밍 훅(청크마다 append)이나 httpx의 stream() 사용 방식으로 처리해야 합니다. 필요하시면 스트리밍 버전도 바로 제공할게요.

  • 훅에서 바디를 읽는 행위가 일부 전송 파이프라인을 바꿀 수 있으니(특히 generator/파일 스트림 전송) 안전한 환경에서 먼저 테스트하세요.

  • 로그/파일에 저장하면 API 키나 민감한 사용자 입력이 유출될 수 있으니 로컬 전용 또는 접근 제한 환경 에서만 저장하세요.

원하시면:

  • messages 필드(프롬프트)만 캡처하고 나머지는 완전 마스킹하는 버전

  • 청크 스트리밍(streaming response)을 로컬 문자열로 합치는 예제

  • 멀티파트(파일 업로드)에서 각 파트 별로 문자열로 저장하는 예제

이 중 어떤 걸 먼저 드릴까요?

response.aread()를 해야 읽을 수 있는거 아니야?

좋은 지적입니다 — 상황에 따라 response.aread()를 await해야만 바디를 읽을 수 있는 경우가 맞습니다. 요점만 먼저 정리하고, 바로 쓸 수 있는 코드 예제(비동기 훅에서 await response.aread() 사용)를 보여드릴게요.

핵심 요약

  1. 동기(httpx.Client) : 훅에서 response.content를 읽으면 내부에서 내용을 가져와 캐시하므로 바로 읽을 수 있습니다.

  2. 비동기(httpx.AsyncClient) : 훅이 async def이면 await response.aread()로 바디(바이트)를 가져와야 안전합니다. (response.content는 동기 속성이라 async 컨텍스트에선 권장되지 않음.)

  3. 주의 — 스트리밍(SSE / chunked / streaming API) : aread()는 전체 바디를 한 번에 읽습니다. 만약 응답이 실시간 스트리밍(예: 서버-전송-이벤트)이라면 aread()로 전체를 읽어버리면 스트리밍 처리를 망칠 수 있습니다. 스트리밍은 response.aiter_bytes() / response.aiter_text()로 청크 단위로 처리하세요.

  4. httpx는 한 번 읽은 바디를 캐시하므로 aread()로 읽은 뒤에도 response.content/response.text로 접근해도 됩니다(대부분의 경우).

바로 쓸 수 있는 비동기 훅 예제 (await response.aread() 사용)

아래는 이전 예제의 response_hook를 await response.aread() 로 안전하게 바꾼 코드입니다.

import json
import httpx
import asyncio
from openai import OpenAI

def redact_headers(headers: dict) -> dict:
    out = {}
    for k, v in headers.items():
        if k.lower() == "authorization" or "api-key" in k.lower():
            out[k] = "REDACTED"
        else:
            out[k] = v
    return out

def redact_text(text: str) -> str:
    # 간단 마스킹 예시
    for sensitive in ("api_key", "apiKey", "key", "authorization", "password"):
        text = text.replace(sensitive + "=", sensitive + "=REDACTED")
    try:
        obj = json.loads(text)
        if isinstance(obj, dict):
            for s in ("api_key", "apiKey", "key", "authorization", "password"):
                if s in obj:
                    obj[s] = "REDACTED"
            return json.dumps(obj, ensure_ascii=False)
    except Exception:
        pass
    return text

async def main():
    captured = {"request": None, "response": None}

    async def request_hook(request: httpx.Request):
        # request.content는 이미 bytes일 가능성 높음
        if request.content is not None:
            try:
                body_str = request.content.decode("utf-8", errors="replace")
            except Exception:
                body_str = repr(request.content)
        else:
            body_str = "<no request.content available or streaming body>"
        captured["request"] = {
            "method": request.method,
            "url": str(request.url),
            "headers": redact_headers(dict(request.headers)),
            "body": redact_text(body_str),
        }

    async def response_hook(response: httpx.Response):
        # 스트리밍 응답인지 간단 체크 (Content-Type 또는 Transfer-Encoding 참고)
        ct = response.headers.get("content-type", "")
        te = response.headers.get("transfer-encoding", "")
        is_streaming = ("text/event-stream" in ct) or ("chunked" in te)

        if is_streaming:
            # 스트리밍은 전체 aread() 하지 않고 청크 단위로 처리해야 함
            chunks = []
            async for chunk in response.aiter_bytes():
                chunks.append(chunk)
            body_bytes = b"".join(chunks)
        else:
            # 일반적인 비동기 응답: 전체 바디를 안전하게 읽음
            body_bytes = await response.aread()

        try:
            body_str = body_bytes.decode("utf-8", errors="replace")
        except Exception:
            body_str = repr(body_bytes)

        # 예쁘게 JSON 포맷 시도
        try:
            parsed = json.loads(body_str)
            body_str = json.dumps(parsed, ensure_ascii=False, indent=2)
        except Exception:
            pass

        captured["response"] = {
            "status_code": response.status_code,
            "headers": redact_headers(dict(response.headers)),
            "body": redact_text(body_str),
        }

    hooks = {"request": [request_hook], "response": [response_hook]}

    async with httpx.AsyncClient(event_hooks=hooks, timeout=60.0) as async_http:
        client = OpenAI(http_client=async_http)
        # OpenAI 호출(비동기)
        await client.chat.completions.create(
            model="gpt-4o-mini",
            messages=[{"role": "user", "content": "안녕하세요 (async aread 예제)"}],
        )

    print("Captured request:", captured["request"]["body"])
    print("Captured response:", captured["response"]["body"][:2000] if captured["response"] else "<none>")

if __name__ == "__main__":
    asyncio.run(main())

추가 팁 / 주의

  • await response.aread()는 응답이 이미 어느 정도 읽혀있지 않다면 전체 바디를 한 번에 읽어 캐시합니다. 대부분의 REST 호출(전체 바디가 작음)에는 이 방법이 편하고 안전합니다.

  • 스트리밍 API(SSE, OpenAI의 일부 스트리밍 인터페이스 등)는 aread()로 읽어버리면 스트림 소비로 인해 원래 의도한 스트리밍 처리(실시간 청크 수신)가 불가능해집니다. 그럴 땐 aiter_bytes()/aiter_text()로 청크를 순회하거나 SDK가 제공하는 스트리밍 콜백을 사용하세요.

  • httpx는 바디를 읽어 캐시에 보관하니, 훅에서 aread()로 읽어도 이후에 response.text/response.json()로 접근 가능하긴 합니다(일반적 동작).

원하시면 지금 바로:

  • 스트리밍(서버-전송-이벤트) 를 청크 단위로 안전하게 캡처해 main() 안의 문자열로 합치는 예제,

  • 혹은 OpenAI SDK의 스트리밍(예:stream=True) 에 특화된 캡처 코드(청크마다 처리해서 단일 문자열로 결합)도 만들어 드릴게요. 어떤 걸 원하시나요?

1. request쪽 한글이 깨져보여. 그리고 여기도 aread로 바꿔줘. 2. body쪽 내용이 모두 하나의 string으로만 나와서 읽기가 힘들어. 그냥 nested json으로 보이는게 읽기 편할 것 같아.

좋은 포인트 두 가지 정리해서 바로 반영했어요.

  • (1) 요청(request) 쪽 한글 깨짐 → Content-Type의 charset을 확인해 정확한 인코딩 으로 디코딩하고, 기본값은 utf-8로 처리합니다. (OpenAI SDK는 JSON 바디를 기본적으로 UTF-8로 보냅니다.)

  • (2) body를 “한 줄짜리 문자열”이 아니라 “중첩 JSON(파이썬 dict)” 으로 저장 → 파싱이 되면 dict/list로 저장하고, 안 되면 텍스트 그대로 둡니다.

또, 응답(response) 은 비동기 환경에서 확실히 await response.aread()로 읽도록 바꿨습니다.
요청(request) 객체는 httpx에서 aread()가 없어서(응답에만 있음) 대신 바이트 바디를 정확한 인코딩으로 디코딩 하는 방식으로 처리합니다. (OpenAI SDK처럼 일반 JSON 요청이면 request.content에서 안전하게 읽힙니다. 진짜 “스트리밍 업로드”를 하신다면 별도 tee 래퍼가 필요합니다—아래에 간단 주석 있습니다.)

업데이트된 비동기 예제 (전역 변수 없이 main() 내부 보관, request 한글/인코딩 개선, response는 aread() 사용, body는 nested JSON으로 저장)

import json
import httpx
import asyncio
from openai import OpenAI

def pick_charset(content_type: str, default: str = "utf-8") -> str:
    # e.g. "application/json; charset=utf-8"
    if not content_type:
        return default
    parts = [p.strip() for p in content_type.split(";")]
    for p in parts[1:]:
        if p.lower().startswith("charset="):
            return p.split("=", 1)[1].strip() or default
    return default

def try_parse_json(text: str):
    try:
        return json.loads(text)
    except Exception:
        return None

def redact_headers(headers: dict) -> dict:
    out = {}
    for k, v in headers.items():
        if k.lower() == "authorization" or "api-key" in k.lower():
            out[k] = "REDACTED"
        else:
            out[k] = v
    return out

def redact_obj(obj):
    # dict/list를 재귀적으로 돌며 민감 키 마스킹
    SENSITIVE_KEYS = {"api_key", "apiKey", "key", "authorization", "password"}
    if isinstance(obj, dict):
        return {k: ("REDACTED" if k in SENSITIVE_KEYS else redact_obj(v)) for k, v in obj.items()}
    if isinstance(obj, list):
        return [redact_obj(x) for x in obj]
    return obj

async def main():
    captured = {"request": None, "response": None}

    async def request_hook(request: httpx.Request):
        # 1) charset 결정
        req_ct = request.headers.get("content-type", "")
        req_charset = pick_charset(req_ct, default="utf-8")

        # 2) 바디 바이트 → 텍스트
        #    (OpenAI SDK의 JSON 호출이라면 request.content가 bytes로 채워져 있습니다)
        if request.content is not None:
            try:
                body_text = request.content.decode(req_charset, errors="replace")
            except Exception:
                # 혹시 모를 인코딩 이슈 대비: 마지막 fallback
                body_text = request.content.decode("utf-8", errors="replace")
        else:
            # 스트리밍 업로드(제너레이터/파일 스트림 등)인 경우 여기에 None이 들어옵니다.
            # 그런 경우, 별도 "tee stream" 래퍼로 `request.stream`을 감싸서 전송되며 캡처해야 합니다.
            body_text = "<streaming or no request.content>"

        # 3) nested JSON으로 저장(가능하면)
        body_obj = try_parse_json(body_text)
        body_saved = redact_obj(body_obj) if body_obj is not None else body_text  # dict/list면 마스킹, 아니면 원문

        captured["request"] = {
            "method": request.method,
            "url": str(request.url),
            "headers": redact_headers(dict(request.headers)),
            # 문자열이 아니라, 가능하면 dict/list로 저장됩니다.
            "body": body_saved,
        }

    async def response_hook(response: httpx.Response):
        # 1) 전체 바디를 비동기로 안전하게 읽음
        body_bytes = await response.aread()

        # 2) charset 결정
        res_ct = response.headers.get("content-type", "")
        res_charset = pick_charset(res_ct, default="utf-8")

        # 3) 바디 바이트 → 텍스트
        try:
            body_text = body_bytes.decode(res_charset, errors="replace")
        except Exception:
            body_text = body_bytes.decode("utf-8", errors="replace")

        # 4) nested JSON으로 저장(가능하면)
        body_obj = try_parse_json(body_text)
        body_saved = redact_obj(body_obj) if body_obj is not None else body_text

        captured["response"] = {
            "status_code": response.status_code,
            "headers": redact_headers(dict(response.headers)),
            "body": body_saved,
        }

    hooks = {"request": [request_hook], "response": [response_hook]}

    async with httpx.AsyncClient(event_hooks=hooks, timeout=60.0) as async_http:
        client = OpenAI(http_client=async_http)

        # 실제 호출(예: chat.completions)
        await client.chat.completions.create(
            model="gpt-4o-mini",
            messages=[{"role": "user", "content": "한글이 제대로 보이는지 확인합니다 😊"}],
        )

    # 여기서 captured["request"]["body"], captured["response"]["body"]는
    # 가능한 경우 "중첩 JSON(dict/list)"로 저장되어 있습니다.
    # (JSON이 아니면 디코딩된 텍스트 문자열)
    print("REQUEST (nested JSON or text):", captured["request"]["body"])
    print("RESPONSE (nested JSON or text):", captured["response"]["body"])

    return captured

if __name__ == "__main__":
    asyncio.run(main())

스트리밍 업로드/다운로드를 쓰는 특수 케이스

  • 요청이 스트리밍 업로드(파일 제너레이터 등)면 request.content가 None일 수 있어요. 이때는 request.stream(비공개 구현일 수 있음)을 tee 래퍼 로 감싸서 전송되는 바이트를 옆에서 모아야 합니다. 필요하시면 tee 래퍼(비동기 제너레이터로 감싸서 bytes를 yield하면서 내부 버퍼에도 append) 코드를 만들어 드릴게요.

  • 응답이 스트리밍(SSE/청크) 이면 await response.aread()로 한 번에 읽지 말고, async for chunk in response.aiter_bytes(): 식으로 청크를 누적 해 nested JSON으로 복원(가능시)하거나, 청크 리스트를 합쳐서 나중에 json.loads()를 시도하세요. 이것도 원하시면 바로 예제 드릴게요.

필요하신 경우:

  • 요청/응답 모두 “반드시 dict로만 보관”하도록 강제하는 버전

  • 특정 키만 남기고 나머지 마스킹(예: messages만 유지)

  • 스트리밍 업/다운로드 tee 래퍼

어떤 걸 더 원하세요?

참고로 body_str을 json.load하고 dump 하니 한글이 제대로 보였었어.

맞아요 — json.loads() → json.dumps() 를 거치면 한글이 “정상 글자”로 보이는 이유가 있습니다. 핵심만 정리하고, 바로 적용 가능한 패치도 드릴게요.

왜 load → dump 하면 한글이 제대로 보일까?

  1. 요청/응답 바이트를 올바른 문자셋(대개 UTF-8) 으로 str로 디코딩해야 합니다.

  2. json.loads()로 파싱하면 파이썬 내부 객체(dict/list) 가 됩니다. 이 상태에서는 한글이 이미 유니코드로 올바르게 보존돼 있어요.

  3. 그런데 json.dumps()의 기본값은 ensure_ascii=True라서, 비 ASCII 문자를\uXXXX로 이스케이프합니다.
    → 이러면 다시 “깨져 보이거나” \uac00 같은 코드값으로 보일 수 있죠.

  4. ensure_ascii=False로 json.dumps(..., ensure_ascii=False, indent=2) 를 쓰면 한글이 그대로 출력됩니다(이스케이프 없이).

즉, “디코딩 → 파싱 → ensure_ascii=False로 직렬화” 순서가 중요합니다.

바로 적용: (1) nested JSON 저장 + (2) 보기 좋게 출력(ensure_ascii=False)

아래는 앞서 드린 비동기 훅 예제에서, nested JSON으로 저장 하고 볼 때는ensure_ascii=False로 덤프하도록 살짝 손 본 버전입니다. (응답은 await response.aread() 유지)

import json
import httpx
import asyncio
from openai import OpenAI
import sys

def pick_charset(content_type: str, default: str = "utf-8") -> str:
    if not content_type:
        return default
    parts = [p.strip() for p in content_type.split(";")]
    for p in parts[1:]:
        if p.lower().startswith("charset="):
            return p.split("=", 1)[1].strip() or default
    return default

def redact_headers(headers: dict) -> dict:
    out = {}
    for k, v in headers.items():
        if k.lower() == "authorization" or "api-key" in k.lower():
            out[k] = "REDACTED"
        else:
            out[k] = v
    return out

def redact_obj(obj):
    SENSITIVE = {"api_key", "apiKey", "key", "authorization", "password"}
    if isinstance(obj, dict):
        return {k: ("REDACTED" if k in SENSITIVE else redact_obj(v)) for k, v in obj.items()}
    if isinstance(obj, list):
        return [redact_obj(x) for x in obj]
    return obj

async def main():
    # (선택) 터미널이 UTF-8이 아닐 수 있으므로 보정
    try:
        sys.stdout.reconfigure(encoding="utf-8")
    except Exception:
        pass

    captured = {"request": None, "response": None}

    async def request_hook(request: httpx.Request):
        req_ct = request.headers.get("content-type", "")
        charset = pick_charset(req_ct, "utf-8")

        if request.content is not None:
            body_text = request.content.decode(charset, errors="replace")
        else:
            body_text = "<streaming or no request.content>"

        # 가능하면 nested JSON(dict/list)로 저장
        try:
            body_obj = json.loads(body_text)
        except Exception:
            body_obj = body_text  # JSON이 아니면 원문 문자열

        captured["request"] = {
            "method": request.method,
            "url": str(request.url),
            "headers": redact_headers(dict(request.headers)),
            "body": redact_obj(body_obj),
        }

    async def response_hook(response: httpx.Response):
        body_bytes = await response.aread()
        res_ct = response.headers.get("content-type", "")
        charset = pick_charset(res_ct, "utf-8")

        body_text = body_bytes.decode(charset, errors="replace")

        try:
            body_obj = json.loads(body_text)
        except Exception:
            body_obj = body_text

        captured["response"] = {
            "status_code": response.status_code,
            "headers": redact_headers(dict(response.headers)),
            "body": redact_obj(body_obj),
        }

    hooks = {"request": [request_hook], "response": [response_hook]}

    async with httpx.AsyncClient(event_hooks=hooks, timeout=60.0) as async_http:
        client = OpenAI(http_client=async_http)

        await client.chat.completions.create(
            model="gpt-4o-mini",
            messages=[{"role": "user", "content": "한글이 잘 보이는지 테스트합니다. 😄"}],
        )

    # === 보기 좋게 출력: ensure_ascii=False + indent ===
    def pretty(x):
        if isinstance(x, (dict, list)):
            return json.dumps(x, ensure_ascii=False, indent=2)
        return x

    print("=== REQUEST ===")
    print(pretty(captured["request"]["body"]))
    print("\n=== RESPONSE ===")
    print(pretty(captured["response"]["body"]))

    return captured

if __name__ == "__main__":
    asyncio.run(main())

포인트

  • 저장 단계 : 가능하면 dict/list 형태로 그대로 저장 합니다(“한 줄 문자열” 대신 중첩 구조).

  • 출력/확인 단계 : json.dumps(..., ensure_ascii=False, indent=2) 로 덤프하면 한글이 그대로 나오고 가독성 도 좋아집니다.

  • 콘솔 인코딩 : Windows 등에서 콘솔 인코딩 문제로 깨질 수 있으니 sys.stdout.reconfigure(encoding="utf-8") 시도(실패해도 무시).

빠른 체크리스트 (step-by-step)

  1. Content-Type에서 charset 확인 (없으면 UTF-8 가정).

  2. 바이트 → 문자열 디코딩.

  3. json.loads()로 객체화(되면 dict/list로 저장).

  4. 출력할 때 json.dumps(..., ensure_ascii=False, indent=2) 사용.

  5. 콘솔 인코딩이 UTF-8인지 확인(필요 시 PYTHONIOENCODING=UTF-8 환경변수 또는 위 코드).

원하시면 요청/응답에서 특정 키만 골라서 pretty-print 하도록 필터링까지 넣어드릴게요. 어떤 키들만 보이면 좋을까요?

Categories:

Updated:

You may also enjoy