Runtime API

import boto3

# ① Bedrock Runtime 클라이언트 생성 — Invoke · Converse 공용
bedrock_runtime = boto3.client(
    'bedrock-runtime',
    region_name='us-east-1'
)

# ② 모델별 네이티브 포맷 그대로 — InvokeModel (PART 2)
response = bedrock_runtime.invoke_model(
    modelId='us.anthropic.claude-sonnet-4-6',
    body=native_json,   # 모델마다 다른 JSON
)

# ③ 모든 모델 통합 인터페이스 — Converse (PART 4)
response = bedrock_runtime.converse(
    modelId='us.anthropic.claude-sonnet-4-6',
    messages=[{'role': 'user', 'content': [{'text': '서울의 인구는?'}]}],
)

• 클라이언트는 하나 — invoke_model()과 converse() 모두 같은 bedrock-runtime
• Invoke — body에 모델별 네이티브 JSON을 그대로 전달
• Converse — messages 통합 시그니처, 모델 무관 동일 코드

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

response = bedrock_runtime.converse(
    modelId='us.anthropic.claude-sonnet-4-6',
    messages=[{'role': 'user', 'content': [
        {'text': '이 회의록을 3줄로 요약해 줘'},
        # 문서 첨부도 content 블록 하나
        {'document': {'format': 'txt',
            'name': 'meeting-notes',
            'source': {'bytes': meeting_notes}}},
    ]}],
    inferenceConfig={'temperature': 0.7,
                     'maxTokens': 512},
)

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

 // end_turn 정상 · tool_use면 함수 호출 루프
 "stopReason": "end_turn",

 // 과금·모니터링 기준
 "usage": {"inputTokens": 1873,
           "outputTokens": 96,
           "totalTokens": 1969}}

response = bedrock_runtime.converse(
    modelId='us.anthropic.claude-sonnet-4-6',
    messages=[
        {
            'role': 'user',
            'content': [{'text': '다음 리스트를 JSON으로 변환해주세요:\n• Apple\n• Banana'}]
        }
    ],
    inferenceConfig={
        'temperature': 0.7,
        'maxTokens': 512,
    }
)

output = response['output']['message']['content'][0]['text']
print(output)
# {"fruits": ["Apple", "Banana"]}

• messages 배열 — role + content 블록 구조는 모델과 무관하게 동일
• inferenceConfig — temperature·maxTokens를 표준 필드로 지정
• 응답 경로 고정 — output.message.content[0].text, 모델 전환 시에도 그대로

response = bedrock_runtime.converse_stream(
    modelId='us.anthropic.claude-sonnet-4-6',
    messages=[{'role': 'user',
               'content': [{'text': '이 회의록을 3줄로 요약해 줘'}]}],
)

# 이벤트 루프 — 통합 스키마라 모델이 바뀌어도 이 코드 그대로
for event in response['stream']:
    if 'contentBlockDelta' in event:
        print(event['contentBlockDelta']['delta']['text'],
              end='', flush=True)
    elif 'messageStop' in event:
        stop_reason = event['messageStop']['stopReason']

• 요청은 동기 호출과 동일 — 메서드만 converse_stream
• contentBlockDelta — 텍스트 델타 · messageStop — stopReason
• usage는 마지막 metadata 이벤트에 — Invoke와 달리 이벤트 이름이 모델 무관

# ① Tool 정의 — toolConfig.tools[].toolSpec 구조
tool_config = {
    "tools": [{
        "toolSpec": {
            "name": "get_weather",
            "description": "도시의 현재 날씨 조회",  # 모델의 선택 근거
            "inputSchema": {"json": {"type": "object",
                "properties": {"city": {"type": "string"}}, "required": ["city"]}},
        }
    }]
}

# ② Converse 호출 — toolConfig로 도구 목록 전달
response = bedrock_runtime.converse(
    modelId='us.anthropic.claude-sonnet-4-6',
    messages=[{'role': 'user', 'content': [{'text': '서울의 날씨는?'}]}],
    toolConfig=tool_config,
)

# ③ 응답 분석 — 모델은 호출 "의도"(name·input)만 JSON으로 반환
for content in response['output']['message']['content']:
    if content.get('toolUse'):
        print(content['toolUse']['name'], content['toolUse']['input'])

• toolSpec — name · description · inputSchema(JSON Schema)로 도구 선언, description이 모델의 선택 근거
• toolConfig — Converse 호출에 도구 목록을 실어 전달
• toolUse — 모델은 호출 의도(name·input)만 반환, 실제 실행은 개발자 몫 — 다음 슬라이드의 ReAct 루프로 이어짐

messages = [{'role': 'user', 'content': [{'text': '서울의 날씨는?'}]}]

while True:
    # ① Reason — 모델이 현재 messages를 보고 다음 행동 결정
    response = bedrock_runtime.converse(
        modelId='us.anthropic.claude-sonnet-4-6',
        messages=messages,
        toolConfig=tool_config)
    messages.append(response['output']['message'])

    # ② 종료 조건 — end_turn이면 최종 응답
    if response['stopReason'] != 'tool_use':
        break

    # ③ Act — 모델이 반환한 도구를 개발자가 실제 실행
    for content in response['output']['message']['content']:
        if content.get('toolUse'):
            tu = content['toolUse']
            result = run_tool(tu['name'], tu['input'])
            # ④ Observe — toolResult를 messages에 추가해 재전달
            messages.append({'role': 'user', 'content': [{
                'toolResult': {'toolUseId': tu['toolUseId'],
                               'content': [{'json': result}]}}]})

• while + stopReason — tool_use면 반복, end_turn이면 break로 종료
• Reason → Act — 모델은 의도만 반환, run_tool() 실행은 개발자 코드
• toolResult (Observe) — toolUseId로 호출과 결과를 짝지어 messages에 재전달

with open('sales-chart.png', 'rb') as f:
    image_bytes = f.read()

response = bedrock_runtime.converse(
    modelId='us.anthropic.claude-sonnet-4-6',
    messages=[{'role': 'user', 'content': [
        {'text': '이 차트의 추세를 두 줄로 설명해줘'},
        {'image': {'format': 'png',
                   'source': {'bytes': image_bytes}}},
    ]}],
)

print(response['output']['message']['content'][0]['text'])

• content 배열에 image 블록 추가 — 텍스트와 같은 층위
• format — png · jpeg · gif · webp
• 문서는 document 블록(PDF·txt·docx 등), 비디오는 video 블록 — 같은 패턴

body = json.dumps({
    'taskType': 'TEXT_IMAGE',
    'textToImageParams': {'text': '노을 지는 서울 스카이라인, 수채화'},
    'imageGenerationConfig': {
        'numberOfImages': 1, 'width': 1024, 'height': 1024,
    },
})

response = bedrock_runtime.invoke_model(
    modelId='amazon.nova-canvas-v1:0', body=body)

output = json.loads(response['body'].read())
image_b64 = output['images'][0]   # base64 — 디코드 후 저장/S3 업로드

• 생성 모델은 Converse 미지원 — InvokeModel 네이티브 포맷 전용
• taskType — TEXT_IMAGE 외 인페인팅·배경 교체 등 과업별 상이
• 프롬프트 구체성이 품질을 가름 — 피사체·스타일·조명까지 명시
• 응답은 base64 이미지 배열 — 비디오(Nova Reel)는 StartAsyncInvoke 비동기