Demonstrates Amazon Bedrock's Converse Streaming API with Claude models and tool use. Includes proper message formatting, tool schema definition, and streaming response handling.

bedrock_converse_stream_tools.py

# MIT-0 License

import boto3
import os

# Initialize Bedrock Runtime client (ensure your AWS credentials are set)
# Get region from environment variable, default to us-east-1
aws_region = os.getenv('AWS_REGION', 'us-east-1')
bedrock = boto3.client('bedrock-runtime', region_name=aws_region)

# Example tool schema
tools = [
    {
        "name": "make_file",
        "description": "Write text to a file",
        "input_schema": {
            "type": "object",
            "properties": {
                "filename": {"type": "string"},
                "lines_of_text": {"type": "array"}
            },
            "required": ["filename", "lines_of_text"]
        }
    }
]

# Prepare messages - content must be a list of content blocks
messages = [
    {
        "role": "user",
        "content": [
            {
                "text": "Can you write a long poem and make a file called poem.txt?"
            }
        ]
    }
]

# Prepare tool configuration using toolSpec
toolConfig = {
    "tools": [
        {
            "toolSpec": {
                "name": tool["name"],
                "description": tool["description"],
                "inputSchema": {
                    "json": tool["input_schema"]
                }
            }
        }
        for tool in tools
    ]
}

# Invoke ConverseStream (streaming API) with fine-grained tool streaming
try:
    response = bedrock.converse_stream(
        modelId='us.anthropic.claude-3-7-sonnet-20250219-v1:0',  # Use Claude 3.7 Sonnet
        messages=messages,
        toolConfig=toolConfig,
        inferenceConfig={
            "maxTokens": 1024
        },
        additionalModelRequestFields={
            "anthropic_beta": ["fine-grained-tool-streaming-2025-05-14"]
        }
    )

    # Process the streaming response
    # Note: With fine-grained tool streaming enabled, you may receive partial/invalid JSON
    # in tool use parameters. Handle carefully and validate on complete reception.
    for event in response['stream']:
        if 'contentBlockDelta' in event:
            delta = event['contentBlockDelta'].get('delta', {})
            if 'text' in delta:
                print(delta['text'], end='', flush=True)
            elif 'toolUseBlock' in delta:
                # Fine-grained tool streaming - may be partial JSON
                print(f"[Tool: {delta['toolUseBlock']}", end='', flush=True)
        elif 'contentBlockStart' in event:
            block = event['contentBlockStart'].get('contentBlock', {})
            block_type = list(block.keys())[0] if block else 'unknown'
            print(f"\n[Block Start: {block_type}]", flush=True)
        elif 'contentBlockStop' in event:
            print(f"[Block Stop]", flush=True)
        elif 'messageStop' in event:
            print("\n[Message Complete]", flush=True)

except Exception as e:
    print(f"Error calling Bedrock: {e}")
    print("\nMake sure you have:")
    print("1. AWS credentials configured")
    print("2. Access to Bedrock in your AWS region")
    print("3. The model 'anthropic.claude-3-5-sonnet-20241022' is available in your region")