에이전트용 HTTP 클라이언트: httpx와 requests

에이전트에게 HTTP 클라이언트가 중요한 이유

requests를 사용한 기본 GET 요청

import requests

url = 'https://api.example.com/data'
response = requests.get(url)

print(response.status_code)  # 200
print(response.text)         # raw string body
print(response.json())       # parsed JSON dict

헤더와 시간 제한 추가하기

import requests

headers = {
    'Authorization': 'Bearer YOUR_API_KEY',
    'Accept': 'application/json'
}

response = requests.get(
    'https://api.example.com/items',
    headers=headers,
    timeout=10  # seconds
)

data = response.json()
print(data)

raise_for_status() — 오류 발생 시 즉시 실패하기

import requests

try:
    response = requests.get('https://api.example.com/missing', timeout=10)
    response.raise_for_status()  # raises if status >= 400
    data = response.json()
except requests.HTTPError as e:
    print(f'HTTP error: {e}')
except requests.RequestException as e:
    print(f'Network error: {e}')

httpx 소개

import httpx

response = httpx.get(
    'https://api.example.com/data',
    headers={'Authorization': 'Bearer YOUR_KEY'},
    timeout=10.0
)

response.raise_for_status()
data = response.json()
print(data)

httpx.AsyncClient를 사용한 비동기 HTTP

import httpx
import asyncio

async def fetch_data(url: str) -> dict:
    async with httpx.AsyncClient(timeout=10.0) as client:
        response = await client.get(
            url,
            headers={'Authorization': 'Bearer YOUR_KEY'}
        )
        response.raise_for_status()
        return response.json()

# result = asyncio.run(fetch_data('https://api.example.com/data'))

여러 비동기 요청을 동시에 보내기

import httpx
import asyncio

async def fetch_all(urls: list) -> list:
    async with httpx.AsyncClient(timeout=10.0) as client:
        tasks = [client.get(url) for url in urls]
        responses = await asyncio.gather(*tasks)
        return [r.json() for r in responses]

urls = [
    'https://api.example.com/item/1',
    'https://api.example.com/item/2',
    'https://api.example.com/item/3'
]
# results = asyncio.run(fetch_all(urls))

세션 재사용과 연결 풀링

import httpx

# Create once, reuse for many requests
client = httpx.Client(
    base_url='https://api.example.com',
    headers={'Authorization': 'Bearer YOUR_KEY'},
    timeout=10.0
)

response1 = client.get('/users')
response2 = client.get('/items')
response3 = client.get('/orders')

client.close()  # always close when done

JSON 본문으로 POST 요청 보내기

import httpx

payload = {
    'query': 'latest AI news',
    'max_results': 5,
    'language': 'en'
}

response = httpx.post(
    'https://api.example.com/search',
    json=payload,
    headers={'Authorization': 'Bearer YOUR_KEY'},
    timeout=15.0
)

response.raise_for_status()
results = response.json()
print(results['items'])

응답 구문 분석하기: text, json, content

import httpx

response = httpx.get('https://api.example.com/report', timeout=10.0)

# For JSON APIs
data = response.json()            # dict or list

# For HTML or plain text
html = response.text              # str

# For binary files
image_bytes = response.content    # bytes

print(type(data), type(html), type(image_bytes))

종합하기: 에이전트 HTTP 가져오기 도구

import httpx
import logging

logger = logging.getLogger(__name__)

def agent_fetch(url: str, headers: dict = None) -> dict:
    try:
        response = httpx.get(
            url,
            headers=headers or {},
            timeout=10.0
        )
        response.raise_for_status()
        logger.info(f'Fetched {url} -> {response.status_code}')
        return {'success': True, 'data': response.json()}
    except httpx.HTTPStatusError as e:
        return {'success': False, 'error': str(e)}
    except httpx.RequestError as e:
        return {'success': False, 'error': f'Network error: {e}'}

지식 확인: HTTP 클라이언트

요약: 에이전트용 HTTP 클라이언트