エージェント向け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クライアント