Invoke Claude 3.7 on Amazon Bedrock

claude37.py

import boto3

def process_stream_response(stream) -> None:
    """Process and print the streaming response from Claude.
    
    Args:
        stream: The response stream from the Bedrock API
    """
    if not stream:
        return
        
    last_was_reasoning = False
    reasoning_complete = False
    
    for event in stream:
        # Check for reasoning content
        if ('contentBlockDelta' in event and 
            'delta' in event['contentBlockDelta'] and 
            'reasoningContent' in event['contentBlockDelta']['delta'] and
            'text' in event['contentBlockDelta']['delta']['reasoningContent']):
            
            print(event['contentBlockDelta']['delta']['reasoningContent']['text'], end="")
            last_was_reasoning = True
        
        # Check for regular content
        elif ('contentBlockDelta' in event and 
              'delta' in event['contentBlockDelta'] and 
              'text' in event['contentBlockDelta']['delta']):
            
            # If we're transitioning from reasoning to regular content
            if last_was_reasoning and not reasoning_complete:
                print("\n\n----- REASONING COMPLETE -----\n\n")
                reasoning_complete = True
                
            print(event['contentBlockDelta']['delta']['text'], end="")
            last_was_reasoning = False


def main():
    """Main function to run the Bedrock conversation."""
    # Initialize client and model
    client = boto3.client('bedrock-runtime', region_name='us-west-2')

    model_id = "us.anthropic.claude-3-7-sonnet-20250219-v1:0"
    
    # Setup conversation
    messages =  [
        {
            'role': 'user',
            'content': [
                {
                    'text': 'How many 5.25 inch Floppy Disks would it take to get to the moon?'
                }
            ]
        }
    ]
    system_message =  [{
            "text": "You are my helpful assistant, helping me understand weird facts and technology bits and bobs"
        }]
    
    # Make API request
    response = client.converse_stream(
        modelId=model_id,
        messages=messages,
        system=system_message,
        inferenceConfig={
            "maxTokens": 4096,
            "temperature": 1,  # must be 1
            # "topP": 1,  # must be unset
        },
        additionalModelRequestFields={
            'thinking': {
                "type": "enabled",
                "budget_tokens": 2000
            }
        }
    )
    
    # Process response
    process_stream_response(response.get('stream'))


if __name__ == '__main__':
    main()