エージェント開発者のためのREST API基礎

HTTPリクエストとは

import requests

# A simple GET request to a public API
response = requests.get('https://api.example.com/users')
print(response.status_code)  # 200
print(response.text)          # raw JSON string

GET — データの取得

import requests

# Fetch users filtered by role
params = {'role': 'admin', 'page': 1, 'limit': 10}
response = requests.get(
    'https://api.example.com/users',
    params=params
)
# URL becomes: /users?role=admin&page=1&limit=10
data = response.json()
print(data['users'])

POST — リソースの作成

import requests
import json

payload = {
    'title': 'Research competitors',
    'assignee': 'agent-001',
    'priority': 'high'
}

response = requests.post(
    'https://api.example.com/tasks',
    json=payload  # sets Content-Type automatically
)
print(response.status_code)  # 201 Created
new_task = response.json()
print('Created task ID:', new_task['id'])

PUTとPATCH — データの更新

import requests

task_id = '42'

# PATCH: only update the status field
response = requests.patch(
    f'https://api.example.com/tasks/{task_id}',
    json={'status': 'completed'}
)
print(response.status_code)  # 200

# PUT: replace the whole task object
full_task = {
    'title': 'Research competitors',
    'assignee': 'agent-001',
    'priority': 'low',
    'status': 'completed'
}
response = requests.put(
    f'https://api.example.com/tasks/{task_id}',
    json=full_task
)
print(response.status_code)  # 200

DELETE — リソースの削除

import requests

task_id = '42'

response = requests.delete(
    f'https://api.example.com/tasks/{task_id}'
)

if response.status_code == 204:
    print('Task deleted successfully')
elif response.status_code == 404:
    print('Task not found — already deleted?')
else:
    print('Unexpected status:', response.status_code)

ステータスコード：2xx 成功

import requests

response = requests.post(
    'https://api.example.com/tasks',
    json={'title': 'New task'}
)

if response.status_code == 201:
    task = response.json()
    print('Created:', task['id'])
elif response.status_code == 200:
    print('Updated existing resource')
else:
    print('Unexpected code:', response.status_code)

ステータスコード：4xx クライアントエラー

import requests

response = requests.get(
    'https://api.example.com/tasks/9999',
    headers={'Authorization': 'Bearer YOUR_KEY'}
)

if response.status_code == 401:
    print('AUTH ERROR: Check your API key')
elif response.status_code == 404:
    print('Task not found')
elif response.status_code == 429:
    retry_after = response.headers.get('Retry-After', 60)
    print(f'Rate limited. Wait {retry_after}s')
elif response.status_code == 400:
    print('Bad request:', response.json().get('error'))

ステータスコード：5xx サーバーエラー

import requests
import time

def get_with_retry(url, headers, max_retries=3):
    for attempt in range(max_retries):
        response = requests.get(url, headers=headers)
        if response.status_code < 500:
            return response  # success or client error
        wait = 2 ** attempt
        print(f'Server error {response.status_code}, retrying in {wait}s...')
        time.sleep(wait)
    return response  # return last response after retries

リクエストヘッダー

import requests

headers = {
    'Content-Type': 'application/json',
    'Authorization': 'Bearer sk-proj-abc123xyz',
    'Accept': 'application/json',
    'User-Agent': 'MyAgent/1.0'
}

response = requests.post(
    'https://api.example.com/analyze',
    headers=headers,
    json={'text': 'Analyze this document'}
)

print(response.json())

JSONリクエストとレスポンスのボディ

import requests

# Send JSON body
response = requests.post(
    'https://api.example.com/summarize',
    json={
        'content': 'Long article text here...',
        'max_length': 150,
        'format': 'bullet_points'
    }
)

# Parse JSON response
result = response.json()

# Always check keys exist
summary = result.get('summary', 'No summary returned')
tokens_used = result.get('usage', {}).get('total_tokens', 0)

print('Summary:', summary)
print('Tokens used:', tokens_used)

すべてを組み合わせる

import requests

class APIClient:
    def __init__(self, base_url, api_key):
        self.base_url = base_url
        self.session = requests.Session()
        self.session.headers.update({
            'Authorization': f'Bearer {api_key}',
            'Content-Type': 'application/json',
            'Accept': 'application/json'
        })

    def get(self, path, params=None):
        r = self.session.get(f'{self.base_url}{path}', params=params)
        r.raise_for_status()
        return r.json()

    def post(self, path, payload):
        r = self.session.post(f'{self.base_url}{path}', json=payload)
        r.raise_for_status()
        return r.json()

# Usage
client = APIClient('https://api.example.com', 'sk-proj-abc123')
tasks = client.get('/tasks', params={'status': 'open'})
new_task = client.post('/tasks', {'title': 'Write report'})

クイックチェック：HTTPメソッド

HTTPの基礎のまとめ