Bedrock Runtime

body = json.dumps({
    # Anthropic 전용 필수
    'anthropic_version': 'bedrock-2023-05-31',
    # 네이티브에선 필수 (Converse는 옵션)
    'max_tokens': 512,
    'messages': [{'role': 'user',
        'content': '이 회의록을 3줄로 요약해 줘'}],
})

response = bedrock_runtime.invoke_model(
    modelId='us.anthropic.claude-sonnet-4-6',
    body=body)

{"content": [{"type": "text",
    "text": "1. 3분기 예산 확정
2. 출시 일정 9월 합의
3. QA 담당 지정"}],

 // 스네이크 케이스 — Converse는 stopReason
 "stop_reason": "end_turn",

 // 이름까지 다름 — Converse는 inputTokens
 "usage": {"input_tokens": 1873,
           "output_tokens": 96}}

import json

# Anthropic Claude - InvokeModel
body = json.dumps({
    "anthropic_version": "bedrock-2023-05-31",
    "max_tokens": 1024,
    "messages": [
        {"role": "user", "content": "Python으로 피보나치 함수 작성"}
    ]
})

response = bedrock_runtime.invoke_model(
    modelId='us.anthropic.claude-sonnet-4-6',
    body=body
)

# 응답 파싱 (Anthropic 포맷)
output = json.loads(response['body'].read())
code = output['content'][0]['text']
print(code)

• body는 모델별 네이티브 JSON 문자열 — anthropic_version 같은 모델 고유 필드 포함
• 응답도 모델별 포맷 — response['body'].read()를 직접 파싱
• 모델을 바꾸면 요청 본문과 파싱 코드를 함께 재작성

response = bedrock_runtime.invoke_model_with_response_stream(
    modelId='us.anthropic.claude-sonnet-4-6',
    body=body   # 동기 호출과 같은 네이티브 JSON
)

# response['body']는 EventStream — chunk 단위로 순회
for event in response['body']:
    chunk = json.loads(event['chunk']['bytes'])

    # Anthropic 스트리밍 포맷 — 타입별 분기
    if chunk['type'] == 'content_block_delta':
        print(chunk['delta']['text'], end='', flush=True)
    elif chunk['type'] == 'message_delta':
        stop_reason = chunk['delta']['stop_reason']

• 요청 body는 동기 호출과 동일 — 메서드만 _with_response_stream
• chunk['bytes'] 안의 델타 JSON도 모델별 포맷 — 타입 이름부터 Anthropic 전용
• content_block_delta에서 텍스트, message_delta에서 stop_reason