AWS API Gateway Streaming Responses: Building Real-Time GenAI Apps Without Workarounds

AWSTemplateFormatVersion: '2010-09-09'
Description: 'API Gateway with Lambda Streaming for GenAI Chatbot'

Resources:
  # REST API
  ChatbotAPI:
    Type: AWS::ApiGateway::RestApi
    Properties:
      Name: GenAI-Chatbot-Streaming-API
      Description: Streaming API for real-time GenAI chatbot responses
      EndpointConfiguration:
        Types:
          - REGIONAL

  # Resource /chat
  ChatResource:
    Type: AWS::ApiGateway::Resource
    Properties:
      RestApiId: !Ref ChatbotAPI
      ParentId: !GetAtt ChatbotAPI.RootResourceId
      PathPart: chat

  # Resource /chat/stream
  StreamResource:
    Type: AWS::ApiGateway::Resource
    Properties:
      RestApiId: !Ref ChatbotAPI
      ParentId: !Ref ChatResource
      PathPart: stream

  # POST method with streaming integration
  StreamMethod:
    Type: AWS::ApiGateway::Method
    Properties:
      RestApiId: !Ref ChatbotAPI
      ResourceId: !Ref StreamResource
      HttpMethod: POST
      AuthorizationType: AWS_IAM  # or COGNITO_USER_POOLS
      Integration:
        Type: AWS  # NOT AWS_PROXY - streaming requires AWS integration
        IntegrationHttpMethod: POST
        Uri: !Sub 'arn:aws:apigateway:${AWS::Region}:lambda:path/2015-03-31/functions/${StreamingLambda.Arn}/invocations'
        # CRITICAL: Enable streaming by setting InvocationType
        IntegrationResponses:
          - StatusCode: 200
            ResponseParameters:
              method.response.header.Content-Type: "'text/event-stream'"
              method.response.header.Cache-Control: "'no-cache'"
              method.response.header.X-Accel-Buffering: "'no'"
        # Enable streaming invocation
        RequestTemplates:
          application/json: |
            {
              "body": $input.json('$'),
              "headers": {
                #foreach($header in $input.params().header.keySet())
                "$header": "$util.escapeJavaScript($input.params().header.get($header))"#if($foreach.hasNext),#end
                #end
              }
            }
      MethodResponses:
        - StatusCode: 200
          ResponseParameters:
            method.response.header.Content-Type: true
            method.response.header.Cache-Control: true
            method.response.header.X-Accel-Buffering: true

  # Lambda function with streaming handler
  StreamingLambda:
    Type: AWS::Lambda::Function
    Properties:
      FunctionName: ChatbotStreamingHandler
      Runtime: nodejs18.x
      Handler: index.handler
      Role: !GetAtt LambdaExecutionRole.Arn
      Timeout: 900  # 15 minutes maximum for streaming
      MemorySize: 1024
      Environment:
        Variables:
          BEDROCK_MODEL_ID: anthropic.claude-3-sonnet-20240229-v1:0
      Code:
        ZipFile: |
          // Placeholder - actual code in next section
          exports.handler = async (event) => {
            return { statusCode: 200, body: 'Streaming handler' };
          };

  # Lambda invoke permission
  LambdaInvokePermission:
    Type: AWS::Lambda::Permission
    Properties:
      FunctionName: !Ref StreamingLambda
      Action: lambda:InvokeFunction
      Principal: apigateway.amazonaws.com
      SourceArn: !Sub 'arn:aws:execute-api:${AWS::Region}:${AWS::AccountId}:${ChatbotAPI}/*'

  # Lambda execution role
  LambdaExecutionRole:
    Type: AWS::IAM::Role
    Properties:
      AssumeRolePolicyDocument:
        Version: '2012-10-17'
        Statement:
          - Effect: Allow
            Principal:
              Service: lambda.amazonaws.com
            Action: sts:AssumeRole
      ManagedPolicyArns:
        - arn:aws:iam::aws:policy/service-role/AWSLambdaBasicExecutionRole
      Policies:
        - PolicyName: BedrockAccess
          PolicyDocument:
            Version: '2012-10-17'
            Statement:
              - Effect: Allow
                Action:
                  - bedrock:InvokeModelWithResponseStream
                Resource: '*'

  # Deployment
  APIDeployment:
    Type: AWS::ApiGateway::Deployment
    DependsOn:
      - StreamMethod
    Properties:
      RestApiId: !Ref ChatbotAPI
      StageName: prod

Outputs:
  APIEndpoint:
    Description: API Gateway endpoint URL
    Value: !Sub 'https://${ChatbotAPI}.execute-api.${AWS::Region}.amazonaws.com/prod'
  StreamingEndpoint:
    Description: Streaming chat endpoint
    Value: !Sub 'https://${ChatbotAPI}.execute-api.${AWS::Region}.amazonaws.com/prod/chat/stream'

import { 
  BedrockRuntimeClient, 
  InvokeModelWithResponseStreamCommand 
} from '@aws-sdk/client-bedrock-runtime';

// Initialize Bedrock client
const bedrockClient = new BedrockRuntimeClient({ region: process.env.AWS_REGION });

/**
 * Lambda streaming handler for GenAI chatbot
 * Uses awslambda.streamifyResponse to enable response streaming
 */
export const handler = awslambda.streamifyResponse(
  async (event, responseStream, _context) => {
    try {
      // Parse incoming request
      const body = JSON.parse(event.body);
      const userMessage = body.message;
      
      if (!userMessage) {
        const error = JSON.stringify({ error: 'Message is required' });
        responseStream.write(error);
        responseStream.end();
        return;
      }

      // Configure metadata for SSE format
      const metadata = {
        statusCode: 200,
        headers: {
          'Content-Type': 'text/event-stream',
          'Cache-Control': 'no-cache',
          'Connection': 'keep-alive',
          'X-Accel-Buffering': 'no' // Disable nginx buffering
        }
      };
      
      // Write metadata first
      responseStream = awslambda.HttpResponseStream.from(
        responseStream, 
        metadata
      );

      // Prepare Bedrock request
      const bedrockRequest = {
        modelId: process.env.BEDROCK_MODEL_ID,
        contentType: 'application/json',
        accept: 'application/json',
        body: JSON.stringify({
          anthropic_version: 'bedrock-2023-05-31',
          max_tokens: 2048,
          messages: [
            {
              role: 'user',
              content: userMessage
            }
          ],
          temperature: 0.7,
          top_p: 0.9
        })
      };

      // Invoke Bedrock with streaming
      const command = new InvokeModelWithResponseStreamCommand(bedrockRequest);
      const response = await bedrockClient.send(command);

      // Process the stream
      let fullResponse = '';
      
      for await (const event of response.body) {
        if (event.chunk) {
          const chunk = JSON.parse(
            Buffer.from(event.chunk.bytes).toString('utf-8')
          );
          
          // Extract token from Claude response
          if (chunk.type === 'content_block_delta') {
            const token = chunk.delta?.text || '';
            fullResponse += token;
            
            // Write as SSE event
            const sseEvent = `data: ${JSON.stringify({ 
              token, 
              done: false 
            })}\n\n`;
            
            responseStream.write(sseEvent);
          }
          
          // Handle completion
          if (chunk.type === 'message_stop') {
            const finalEvent = `data: ${JSON.stringify({ 
              token: '', 
              done: true,
              fullResponse 
            })}\n\n`;
            
            responseStream.write(finalEvent);
          }
        }
      }

      // Close the stream
      responseStream.end();
      
    } catch (error) {
      console.error('Streaming error:', error);
      
      const errorEvent = `data: ${JSON.stringify({ 
        error: error.message,
        done: true 
      })}\n\n`;
      
      responseStream.write(errorEvent);
      responseStream.end();
    }
  }
);

/**
 * Performance notes:
 * - Time-to-first-token: ~300-500ms (vs 3+ seconds buffered)
 * - Total streaming time: Depends on response length
 * - Memory usage: Constant (no buffering required)
 * - Concurrent connections: Monitor Lambda concurrency limits
 */

import json
import os
import boto3
from typing import Iterator

# Initialize Bedrock client
bedrock_runtime = boto3.client('bedrock-runtime', region_name=os.environ['AWS_REGION'])

def handler(event, context):
    """
    Lambda streaming handler for GenAI chatbot (Python)
    Returns an iterator that yields response chunks
    """
    
    def generate_response() -> Iterator[str]:
        """Generator function that yields SSE-formatted chunks"""
        
        try:
            # Parse request
            body = json.loads(event['body'])
            user_message = body.get('message')
            
            if not user_message:
                yield f'data: {json.dumps({"error": "Message required", "done": True})}\n\n'
                return
            
            # Prepare Bedrock request
            bedrock_request = {
                'modelId': os.environ['BEDROCK_MODEL_ID'],
                'contentType': 'application/json',
                'accept': 'application/json',
                'body': json.dumps({
                    'anthropic_version': 'bedrock-2023-05-31',
                    'max_tokens': 2048,
                    'messages': [
                        {
                            'role': 'user',
                            'content': user_message
                        }
                    ],
                    'temperature': 0.7,
                    'top_p': 0.9
                })
            }
            
            # Invoke Bedrock with streaming
            response = bedrock_runtime.invoke_model_with_response_stream(**bedrock_request)
            
            # Process stream
            full_response = ''
            
            for event_chunk in response['body']:
                chunk = event_chunk.get('chunk')
                
                if chunk:
                    chunk_data = json.loads(chunk['bytes'].decode('utf-8'))
                    
                    # Extract token
                    if chunk_data['type'] == 'content_block_delta':
                        token = chunk_data.get('delta', {}).get('text', '')
                        full_response += token
                        
                        # Yield SSE event
                        sse_event = f'data: {json.dumps({"token": token, "done": False})}\n\n'
                        yield sse_event
                    
                    # Handle completion
                    if chunk_data['type'] == 'message_stop':
                        final_event = f'data: {json.dumps({"token": "", "done": True, "fullResponse": full_response})}\n\n'
                        yield final_event
        
        except Exception as e:
            error_event = f'data: {json.dumps({"error": str(e), "done": True})}\n\n'
            yield error_event
    
    # Return streaming response configuration
    return {
        'statusCode': 200,
        'headers': {
            'Content-Type': 'text/event-stream',
            'Cache-Control': 'no-cache',
            'Connection': 'keep-alive',
            'X-Accel-Buffering': 'no'
        },
        'body': generate_response()  # Return generator for streaming
    }

import React, { useState, useRef } from 'react';

const ChatComponent = () => {
  const [message, setMessage] = useState('');
  const [response, setResponse] = useState('');
  const [isStreaming, setIsStreaming] = useState(false);
  const eventSourceRef = useRef(null);

  const sendMessage = async () => {
    if (!message.trim()) return;
    
    setIsStreaming(true);
    setResponse('');
    
    try {
      // Use fetch with EventSource-compatible endpoint
      const apiUrl = 'https://your-api-id.execute-api.us-east-1.amazonaws.com/prod/chat/stream';
      
      const response = await fetch(apiUrl, {
        method: 'POST',
        headers: {
          'Content-Type': 'application/json',
          // Add authentication headers as needed
          'Authorization': `Bearer ${yourAuthToken}`
        },
        body: JSON.stringify({ message })
      });

      // Read streaming response
      const reader = response.body.getReader();
      const decoder = new TextDecoder();
      
      let buffer = '';
      
      while (true) {
        const { done, value } = await reader.read();
        
        if (done) break;
        
        // Decode chunk
        buffer += decoder.decode(value, { stream: true });
        
        // Process SSE events
        const events = buffer.split('\n\n');
        buffer = events.pop(); // Keep incomplete event in buffer
        
        for (const event of events) {
          if (event.startsWith('data: ')) {
            const data = JSON.parse(event.slice(6));
            
            if (data.error) {
              console.error('Streaming error:', data.error);
              setIsStreaming(false);
              return;
            }
            
            if (!data.done) {
              // Append token to response
              setResponse(prev => prev + data.token);
            } else {
              // Stream complete
              console.log('Full response:', data.fullResponse);
              setIsStreaming(false);
            }
          }
        }
      }
      
    } catch (error) {
      console.error('Error:', error);
      setResponse('Error: ' + error.message);
      setIsStreaming(false);
    }
  };

  return (
    

      

        {response || 'Response will appear here...'}
        {isStreaming && ▋}
      
      
      

         setMessage(e.target.value)}
          onKeyPress={(e) => e.key === 'Enter' && sendMessage()}
          placeholder="Ask a question..."
          disabled={isStreaming}
        />
        
          {isStreaming ? 'Streaming...' : 'Send'}
        
      
    
  );
};

export default ChatComponent;

/**
 * Performance metrics (typical):
 * - Time-to-first-token: 300-500ms (was 3+ seconds)
 * - Perceived latency improvement: 85%
 * - User engagement: +40% (faster responses feel more natural)
 */

import { CloudWatchClient, GetMetricStatisticsCommand } from '@aws-sdk/client-cloudwatch';

const cloudwatch = new CloudWatchClient({ region: process.env.AWS_REGION });

export const handler = awslambda.streamifyResponse(
  async (event, responseStream, _context) => {
    const metadata = {
      statusCode: 200,
      headers: {
        'Content-Type': 'text/event-stream',
        'Cache-Control': 'no-cache',
        'Connection': 'keep-alive',
        'Access-Control-Allow-Origin': '*'
      }
    };
    
    responseStream = awslambda.HttpResponseStream.from(responseStream, metadata);
    
    let eventId = 0;
    const interval = 5000; // Update every 5 seconds
    const maxDuration = 14 * 60 * 1000; // 14 minutes (under 15min limit)
    const startTime = Date.now();
    
    try {
      while (Date.now() - startTime < maxDuration) {
        // Fetch metrics from CloudWatch
        const metrics = await getCloudWatchMetrics();
        
        // Send SSE event with ID for resumption
        const sseEvent = [
          `id: ${eventId++}`,
          `event: metrics`,
          `data: ${JSON.stringify(metrics)}`,
          '',
          ''
        ].join('\n');
        
        responseStream.write(sseEvent);
        
        // Heartbeat to keep connection alive
        if (eventId % 6 === 0) { // Every 30 seconds
          responseStream.write(': heartbeat\n\n');
        }
        
        // Wait for next interval
        await new Promise(resolve => setTimeout(resolve, interval));
      }
      
      // Graceful shutdown after 14 minutes
      responseStream.write('event: close\ndata: {"message": "Stream timeout"}\n\n');
      
    } catch (error) {
      const errorEvent = `event: error\ndata: ${JSON.stringify({ error: error.message })}\n\n`;
      responseStream.write(errorEvent);
    }
    
    responseStream.end();
  }
);

async function getCloudWatchMetrics() {
  const endTime = new Date();
  const startTime = new Date(endTime.getTime() - 5 * 60 * 1000); // Last 5 minutes
  
  const command = new GetMetricStatisticsCommand({
    Namespace: 'AWS/Lambda',
    MetricName: 'Invocations',
    Dimensions: [
      {
        Name: 'FunctionName',
        Value: process.env.FUNCTION_NAME
      }
    ],
    StartTime: startTime,
    EndTime: endTime,
    Period: 60,
    Statistics: ['Sum', 'Average']
  });
  
  const response = await cloudwatch.send(command);
  
  return {
    timestamp: new Date().toISOString(),
    invocations: response.Datapoints?.[0]?.Sum || 0,
    average: response.Datapoints?.[0]?.Average || 0
  };
}

import React, { useEffect, useState, useRef } from 'react';

const DashboardClient = () => {
  const [metrics, setMetrics] = useState(null);
  const [connectionStatus, setConnectionStatus] = useState('disconnected');
  const eventSourceRef = useRef(null);
  const lastEventIdRef = useRef(null);

  useEffect(() => {
    connectSSE();
    
    return () => {
      if (eventSourceRef.current) {
        eventSourceRef.current.close();
      }
    };
  }, []);

  const connectSSE = () => {
    // Build URL with Last-Event-ID if reconnecting
    let url = 'https://your-api-id.execute-api.us-east-1.amazonaws.com/prod/metrics/stream';
    
    if (lastEventIdRef.current) {
      url += `?lastEventId=${lastEventIdRef.current}`;
    }

    const eventSource = new EventSource(url);
    eventSourceRef.current = eventSource;

    eventSource.onopen = () => {
      console.log('SSE connection opened');
      setConnectionStatus('connected');
    };

    // Listen for 'metrics' events
    eventSource.addEventListener('metrics', (event) => {
      lastEventIdRef.current = event.lastEventId;
      const data = JSON.parse(event.data);
      setMetrics(data);
    });

    // Listen for 'close' events
    eventSource.addEventListener('close', (event) => {
      console.log('Server closed stream:', event.data);
      eventSource.close();
      setConnectionStatus('closed');
      
      // Reconnect after 5 seconds
      setTimeout(() => {
        console.log('Reconnecting...');
        connectSSE();
      }, 5000);
    });

    eventSource.onerror = (error) => {
      console.error('SSE error:', error);
      setConnectionStatus('error');
      eventSource.close();
      
      // Automatic reconnection with exponential backoff
      const backoff = Math.min(30000, 1000 * Math.pow(2, reconnectAttempts));
      setTimeout(() => {
        console.log(`Reconnecting in ${backoff/1000}s...`);
        connectSSE();
      }, backoff);
    };
  };

  return (
    

      

        Status: {connectionStatus}
      
      
      {metrics && (
        

          
Live Metrics
          
Timestamp: {metrics.timestamp}
          
Invocations: {metrics.invocations}
          
Average: {metrics.average.toFixed(2)}
        
      )}
    
  );
};

export default DashboardClient;

import { S3Client, GetObjectCommand } from '@aws-sdk/client-s3';

const s3Client = new S3Client({ region: process.env.AWS_REGION });

export const handler = awslambda.streamifyResponse(
  async (event, responseStream, _context) => {
    try {
      // Extract file key from path parameters
      const fileKey = event.pathParameters?.fileKey;
      const bucket = process.env.BUCKET_NAME;
      
      if (!fileKey) {
        const metadata = { statusCode: 400, headers: {} };
        responseStream = awslambda.HttpResponseStream.from(responseStream, metadata);
        responseStream.write('File key required');
        responseStream.end();
        return;
      }

      // Get S3 object
      const command = new GetObjectCommand({
        Bucket: bucket,
        Key: fileKey
      });
      
      const s3Response = await s3Client.send(command);
      
      // Set appropriate headers
      const metadata = {
        statusCode: 200,
        headers: {
          'Content-Type': s3Response.ContentType || 'application/octet-stream',
          'Content-Length': s3Response.ContentLength.toString(),
          'Content-Disposition': `attachment; filename="${fileKey.split('/').pop()}"`,
          'Cache-Control': 'private, max-age=3600'
        }
      };
      
      responseStream = awslambda.HttpResponseStream.from(responseStream, metadata);
      
      // Stream S3 object directly to response
      // IMPORTANT: Use streams to avoid loading entire file into memory
      const readable = s3Response.Body;
      
      for await (const chunk of readable) {
        responseStream.write(chunk);
      }
      
      responseStream.end();
      
    } catch (error) {
      console.error('File streaming error:', error);
      
      const metadata = {
        statusCode: error.name === 'NoSuchKey' ? 404 : 500,
        headers: { 'Content-Type': 'application/json' }
      };
      
      responseStream = awslambda.HttpResponseStream.from(responseStream, metadata);
      responseStream.write(JSON.stringify({ error: error.message }));
      responseStream.end();
    }
  }
);

/**
 * Performance comparison:
 * 
 * Pre-signed URL approach:
 * - Client makes API request
 * - Lambda generates pre-signed URL (~50ms)
 * - Client makes second request to S3 (~100ms)
 * - Total: ~150ms + 2 round trips
 * 
 * Streaming approach:
 * - Client makes API request
 * - Lambda streams S3 object directly
 * - Total: ~50ms + 1 round trip
 * 
 * Benefits:
 * - 50% latency reduction
 * - Consistent authentication (API Gateway layer)
 * - No exposed S3 bucket structure
 * - Simpler client implementation
 */