event-driven-serverless-systems

AWS サーバーレスとイベント駆動型アーキテクチャ

このスキルでは、Well-Architected Framework の原則に基づいて、AWS 上でサーバーレスアプリケーションとイベント駆動型アーキテクチャを構築するための包括的なガイダンスを提供します。

AWS ドキュメントに関する要件

重要：このスキルでは、正確で最新の AWS 情報を得るために AWS MCP ツールが必要です。

AWS に関する質問に回答する前に

1. 必ず AWS MCP ツールで確認してください（利用可能な場合）：

   • mcp__aws-mcp__aws___search_documentation または mcp__*awsdocs*__aws___search_documentation - AWS ドキュメントを検索します
   • mcp__aws-mcp__aws___read_documentation または mcp__*awsdocs*__aws___read_documentation - 特定のページを読みます
   • mcp__aws-mcp__aws___get_regional_availability - サービスの提供状況を確認します

2. AWS MCP ツールを利用できない場合：

   • AWS MCP の設定を案内してください：AWS MCP 設定ガイドを参照してください
   • ユーザーの環境に適したオプションを判断できるよう支援してください：
     • uvx と AWS 認証情報がある → フル AWS MCP サーバー
     • Python／認証情報がない → AWS ドキュメント MCP（認証不要）
   • 判断できない場合 → どのオプションを使うかユーザーに尋ねてください

サーバーレス MCP サーバー

このスキルでは、サーバーレスに特化した MCP サーバーを活用して、開発ワークフローを強化できます。

AWS Serverless MCP Server

目的：SAM CLI を使用したサーバーレスアプリケーションのライフサイクル全体の管理

• 新しいサーバーレスアプリケーションを初期化する
• サーバーレスアプリケーションをデプロイする
• Lambda 関数をローカルでテストする
• SAM テンプレートを生成する
• サーバーレスアプリケーションのライフサイクルを管理する

AWS Lambda Tool MCP Server

目的：Lambda 関数をツールとして実行する

• Lambda 関数を直接呼び出す
• Lambda の統合をテストする
• プライベートリソースへのアクセスが必要なワークフローを実行する
• Lambda ベースの自動化を実行する

AWS Step Functions MCP Server

目的：複雑なワークフローとオーケストレーションを実行する

• ステートマシンを作成・管理する
• ワークフローのオーケストレーションを実行する
• 分散トランザクションを処理する
• Saga パターンを実装する
• マイクロサービスを連携させる

Amazon SNS/SQS MCP Server

目的：イベント駆動型メッセージングとキュー管理

• SNS トピックにメッセージを発行する
• SQS キューからメッセージを送受信する
• イベント駆動型の通信を管理する
• パブリッシュ／サブスクライブパターンを実装する
• 非同期処理を行う

このスキルを使用する場面

次のような場合に、このスキルを使用してください：

• Lambda を使用したサーバーレスアプリケーションを構築する
• イベント駆動型アーキテクチャを設計する
• マイクロサービスパターンを実装する
• 非同期処理のワークフローを作成する
• 複数サービスにまたがるトランザクションをオーケストレーションする
• リアルタイムデータ処理パイプラインを構築する
• 分散トランザクション向けに Saga パターンを実装する
• スケーラビリティと耐障害性を考慮して設計する

AWS Well-Architected のサーバーレス設計原則

1. 高速、シンプル、単一目的

関数は簡潔で、単一の目的を持つようにします

// ✅ GOOD - Single purpose, focused function
export const processOrder = async (event: OrderEvent) => {
  // Only handles order processing
  const order = await validateOrder(event);
  await saveOrder(order);
  await publishOrderCreatedEvent(order);
  return { statusCode: 200, body: JSON.stringify({ orderId: order.id }) };
};

// ❌ BAD - Function does too much
export const handleEverything = async (event: any) => {
  // Handles orders, inventory, payments, shipping...
  // Too many responsibilities
};

関数の環境効率を維持し、コストを意識します：

• コールドスタート時間を最小限に抑える
• メモリ割り当てを最適化する
• 必要な場合にのみプロビジョンドコンカレンシーを使用する
• 接続の再利用を活用する

2. リクエストの総数ではなく、同時リクエストを考える

ボリュームではなく、同時実行性を考慮して設計します

Lambda は水平スケーリングするため、次の点に重点を置いて設計する必要があります：

• 同時実行数の上限
• 下流サービスのスロットリング
• 共有リソースの競合
• コネクションプールのサイズ

// Consider concurrent Lambda executions accessing DynamoDB
const table = new dynamodb.Table(this, 'Table', {
  billingMode: dynamodb.BillingMode.PAY_PER_REQUEST, // Auto-scales with load
});

// Or with provisioned capacity + auto-scaling
const table = new dynamodb.Table(this, 'Table', {
  billingMode: dynamodb.BillingMode.PROVISIONED,
  readCapacity: 5,
  writeCapacity: 5,
});

// Enable auto-scaling for concurrent load
table.autoScaleReadCapacity({ minCapacity: 5, maxCapacity: 100 });
table.autoScaleWriteCapacity({ minCapacity: 5, maxCapacity: 100 });

3. 何も共有しない

関数のランタイム環境は短命です

// ❌ BAD - Relying on local file system
export const handler = async (event: any) => {
  fs.writeFileSync('/tmp/data.json', JSON.stringify(data)); // Lost after execution
};

// ✅ GOOD - Use persistent storage
export const handler = async (event: any) => {
  await s3.putObject({
    Bucket: process.env.BUCKET_NAME,
    Key: 'data.json',
    Body: JSON.stringify(data),
  });
};

状態管理：

• 永続状態には DynamoDB を使用する
• ワークフローの状態には Step Functions を使用する
• セッション状態には ElastiCache を使用する
• ファイルストレージには S3 を使用する

4. ハードウェアへの依存を前提としない

アプリケーションはハードウェアに依存しないようにします

インフラストラクチャは予告なく変更される可能性があります：

• Lambda 関数は異なるハードウェア上で実行される場合がある
• コンテナインスタンスは置き換えられる場合がある
• 基盤となるインフラストラクチャを前提としない

移植性を考慮して設計します：

• 設定には環境変数を使用する
• ハードウェア固有の最適化を避ける
• 異なる環境でテストする

5. 関数の連鎖ではなく、ステートマシンでオーケストレーションする

オーケストレーションには Step Functions を使用します

// ❌ BAD - Lambda function chaining
export const handler1 = async (event: any) => {
  const result = await processStep1(event);
  await lambda.invoke({
    FunctionName: 'handler2',
    Payload: JSON.stringify(result),
  });
};

// ✅ GOOD - Step Functions orchestration
const stateMachine = new stepfunctions.StateMachine(this, 'OrderWorkflow', {
  definition: stepfunctions.Chain
    .start(validateOrder)
    .next(processPayment)
    .next(shipOrder)
    .next(sendConfirmation),
});

Step Functions のメリット：

• ワークフローを視覚的に表現できます
• エラー処理と再試行が組み込まれています
• 実行履歴を確認してデバッグできます
• 並列実行と逐次実行に対応しています
• コードを書かずにサービスを統合できます

6. イベントを使ってトランザクションをトリガーする

同期的なリクエスト／レスポンスではなく、イベント駆動にします

// Pattern: Event-driven processing
const bucket = new s3.Bucket(this, 'DataBucket');

bucket.addEventNotification(
  s3.EventType.OBJECT_CREATED,
  new s3n.LambdaDestination(processFunction),
  { prefix: 'uploads/' }
);

// Pattern: EventBridge integration
const rule = new events.Rule(this, 'OrderRule', {
  eventPattern: {
    source: ['orders'],
    detailType: ['OrderPlaced'],
  },
});

rule.addTarget(new targets.LambdaFunction(processOrderFunction));

メリット：

• サービス間の疎結合を実現できます
• 非同期処理が可能です
• フォールトトレランスが向上します
• 独立してスケーリングできます

7. 障害と重複を想定して設計する

操作は冪等である必要があります

// ✅ GOOD - Idempotent operation
export const handler = async (event: SQSEvent) => {
  for (const record of event.Records) {
    const orderId = JSON.parse(record.body).orderId;

    // Check if already processed (idempotency)
    const existing = await dynamodb.getItem({
      TableName: process.env.TABLE_NAME,
      Key: { orderId },
    });

    if (existing.Item) {
      console.log('Order already processed:', orderId);
      continue; // Skip duplicate
    }

    // Process order
    await processOrder(orderId);

    // Mark as processed
    await dynamodb.putItem({
      TableName: process.env.TABLE_NAME,
      Item: { orderId, processedAt: Date.now() },
    });
  }
};

指数バックオフを使った再試行ロジックを実装します：

async function withRetry<T>(fn: () => Promise<T>, maxRetries = 3): Promise<T> {
  for (let i = 0; i < maxRetries; i++) {
    try {
      return await fn();
    } catch (error) {
      if (i === maxRetries - 1) throw error;
      await new Promise(resolve => setTimeout(resolve, Math.pow(2, i) * 1000));
    }
  }
  throw new Error('Max retries exceeded');
}

イベント駆動アーキテクチャのパターン

パターン 1：イベントルーター（EventBridge）

イベントのルーティングとフィルタリングに EventBridge を使用します：

// Create custom event bus
const eventBus = new events.EventBus(this, 'AppEventBus', {
  eventBusName: 'application-events',
});

// Define event schema
const schema = new events.Schema(this, 'OrderSchema', {
  schemaName: 'OrderPlaced',
  definition: events.SchemaDefinition.fromInline({
    openapi: '3.0.0',
    info: { version: '1.0.0', title: 'Order Events' },
    paths: {},
    components: {
      schemas: {
        OrderPlaced: {
          type: 'object',
          properties: {
            orderId: { type: 'string' },
            customerId: { type: 'string' },
            amount: { type: 'number' },
          },
        },
      },
    },
  }),
});

// Create rules for different consumers
new events.Rule(this, 'ProcessOrderRule', {
  eventBus,
  eventPattern: {
    source: ['orders'],
    detailType: ['OrderPlaced'],
  },
  targets: [new targets.LambdaFunction(processOrderFunction)],
});

new events.Rule(this, 'NotifyCustomerRule', {
  eventBus,
  eventPattern: {
    source: ['orders'],
    detailType: ['OrderPlaced'],
  },
  targets: [new targets.LambdaFunction(notifyCustomerFunction)],
});

パターン 2：キューベースの処理（SQS）

信頼性の高い非同期処理に SQS を使用します：

// Standard queue for at-least-once delivery
const queue = new sqs.Queue(this, 'ProcessingQueue', {
  visibilityTimeout: Duration.seconds(300),
  retentionPeriod: Duration.days(14),
  deadLetterQueue: {
    queue: dlq,
    maxReceiveCount: 3,
  },
});

// FIFO queue for ordered processing
const fifoQueue = new sqs.Queue(this, 'OrderedQueue', {
  fifo: true,
  contentBasedDeduplication: true,
  deduplicationScope: sqs.DeduplicationScope.MESSAGE_GROUP,
});

// Lambda consumer
new lambda.EventSourceMapping(this, 'QueueConsumer', {
  target: processingFunction,
  eventSourceArn: queue.queueArn,
  batchSize: 10,
  maxBatchingWindow: Duration.seconds(5),
});

パターン 3：Pub/Sub（SNS + SQS ファンアウト）

複数のコンシューマー向けにファンアウトパターンを実装します：

// Create SNS topic
const topic = new sns.Topic(this, 'OrderTopic', {
  displayName: 'Order Events',
});

// Multiple SQS queues subscribe to topic
const inventoryQueue = new sqs.Queue(this, 'InventoryQueue');
const shippingQueue = new sqs.Queue(this, 'ShippingQueue');
const analyticsQueue = new sqs.Queue(this, 'AnalyticsQueue');

topic.addSubscription(new subscriptions.SqsSubscription(inventoryQueue));
topic.addSubscription(new subscriptions.SqsSubscription(shippingQueue));
topic.addSubscription(new subscriptions.SqsSubscription(analyticsQueue));

// Each queue has its own Lambda consumer
new lambda.EventSourceMapping(this, 'InventoryConsumer', {
  target: inventoryFunction,
  eventSourceArn: inventoryQueue.queueArn,
});

パターン 4：Step Functions による Saga パターン

分散トランザクションを実装します：

const reserveFlight = new tasks.LambdaInvoke(this, 'ReserveFlight', {
  lambdaFunction: reserveFlightFunction,
  outputPath: '$.Payload',
});

const reserveHotel = new tasks.LambdaInvoke(this, 'ReserveHotel', {
  lambdaFunction: reserveHotelFunction,
  outputPath: '$.Payload',
});

const processPayment = new tasks.LambdaInvoke(this, 'ProcessPayment', {
  lambdaFunction: processPaymentFunction,
  outputPath: '$.Payload',
});

// Compensating transactions
const cancelFlight = new tasks.LambdaInvoke(this, 'CancelFlight', {
  lambdaFunction: cancelFlightFunction,
});

const cancelHotel = new tasks.LambdaInvoke(this, 'CancelHotel', {
  lambdaFunction: cancelHotelFunction,
});

// Define saga with compensation
const definition = reserveFlight
  .next(reserveHotel)
  .next(processPayment)
  .addCatch(cancelHotel.next(cancelFlight), {
    resultPath: '$.error',
  });

new stepfunctions.StateMachine(this, 'BookingStateMachine', {
  definition,
  timeout: Duration.minutes(5),
});

パターン 5：イベントソーシング

イベントを信頼できる唯一の情報源として保存します：

// Event store with DynamoDB
const eventStore = new dynamodb.Table(this, 'EventStore', {
  partitionKey: { name: 'aggregateId', type: dynamodb.AttributeType.STRING },
  sortKey: { name: 'version', type: dynamodb.AttributeType.NUMBER },
  stream: dynamodb.StreamViewType.NEW_IMAGE,
});

// Lambda function stores events
export const handleCommand = async (event: any) => {
  const { aggregateId, eventType, eventData } = event;

  // Get current version
  const items = await dynamodb.query({
    TableName: process.env.EVENT_STORE,
    KeyConditionExpression: 'aggregateId = :id',
    ExpressionAttributeValues: { ':id': aggregateId },
    ScanIndexForward: false,
    Limit: 1,
  });

  const nextVersion = items.Items?.[0]?.version + 1 || 1;

  // Append new event
  await dynamodb.putItem({
    TableName: process.env.EVENT_STORE,
    Item: {
      aggregateId,
      version: nextVersion,
      eventType,
      eventData,
      timestamp: Date.now(),
    },
  });
};

// Projections read from event stream
eventStore.grantStreamRead(projectionFunction);

サーバーレスアーキテクチャのパターン

パターン 1：API 駆動のマイクロサービス

Lambda をバックエンドとする REST API：

const api = new apigateway.RestApi(this, 'Api', {
  restApiName: 'microservices-api',
  deployOptions: {
    throttlingRateLimit: 1000,
    throttlingBurstLimit: 2000,
    tracingEnabled: true,
  },
});

// User service
const users = api.root.addResource('users');
users.addMethod('GET', new apigateway.LambdaIntegration(getUsersFunction));
users.addMethod('POST', new apigateway.LambdaIntegration(createUserFunction));

// Order service
const orders = api.root.addResource('orders');
orders.addMethod('GET', new apigateway.LambdaIntegration(getOrdersFunction));
orders.addMethod('POST', new apigateway.LambdaIntegration(createOrderFunction));

パターン 2：ストリーム処理

Kinesis によるリアルタイムデータ処理：

const stream = new kinesis.Stream(this, 'DataStream', {
  shardCount: 2,
  retentionPeriod: Duration.days(7),
});

// Lambda processes stream records
new lambda.EventSourceMapping(this, 'StreamProcessor', {
  target: processFunction,
  eventSourceArn: stream.streamArn,
  batchSize: 100,
  maxBatchingWindow: Duration.seconds(5),
  parallelizationFactor: 10,
  startingPosition: lambda.StartingPosition.LATEST,
  retryAttempts: 3,
  bisectBatchOnError: true,
  onFailure: new lambdaDestinations.SqsDestination(dlq),
});

パターン 3：非同期タスク処理

バックグラウンドジョブの処理：

// SQS queue for tasks
const taskQueue = new sqs.Queue(this, 'TaskQueue', {
  visibilityTimeout: Duration.minutes(5),
  receiveMessageWaitTime: Duration.seconds(20), // Long polling
  deadLetterQueue: {
    queue: dlq,
    maxReceiveCount: 3,
  },
});

// Lambda worker processes tasks
const worker = new lambda.Function(this, 'TaskWorker', {
  // ... configuration
  reservedConcurrentExecutions: 10, // Control concurrency
});

new lambda.EventSourceMapping(this, 'TaskConsumer', {
  target: worker,
  eventSourceArn: taskQueue.queueArn,
  batchSize: 10,
  reportBatchItemFailures: true, // Partial batch failure handling
});

パターン 4：スケジュールジョブ

EventBridge による定期的な処理：

// Daily cleanup job
new events.Rule(this, 'DailyCleanup', {
  schedule: events.Schedule.cron({ hour: '2', minute: '0' }),
  targets: [new targets.LambdaFunction(cleanupFunction)],
});

// Process every 5 minutes
new events.Rule(this, 'FrequentProcessing', {
  schedule: events.Schedule.rate(Duration.minutes(5)),
  targets: [new targets.LambdaFunction(processFunction)],
});

パターン 5：Webhook 処理

外部 Webhook を処理します：

// API Gateway endpoint for webhooks
const webhookApi = new apigateway.RestApi(this, 'WebhookApi', {
  restApiName: 'webhooks',
});

const webhook = webhookApi.root.addResource('webhook');
webhook.addMethod('POST', new apigateway.LambdaIntegration(webhookFunction, {
  proxy: true,
  timeout: Duration.seconds(29), // API Gateway max
}));

// Lambda handler validates and queues webhook
export const handler = async (event: APIGatewayProxyEvent) => {
  // Validate webhook signature
  const isValid = validateSignature(event.headers, event.body);
  if (!isValid) {
    return { statusCode: 401, body: 'Invalid signature' };
  }

  // Queue for async processing
  await sqs.sendMessage({
    QueueUrl: process.env.QUEUE_URL,
    MessageBody: event.body,
  });

  // Return immediately
  return { statusCode: 202, body: 'Accepted' };
};

ベストプラクティス

エラー処理

包括的なエラー処理を実装します：

export const handler = async (event: SQSEvent) => {
  const failures: SQSBatchItemFailure[] = [];

  for (const record of event.Records) {
    try {
      await processRecord(record);
    } catch (error) {
      console.error('Failed to process record:', record.messageId, error);
      failures.push({ itemIdentifier: record.messageId });
    }
  }

  // Return partial batch failures for retry
  return { batchItemFailures: failures };
};

デッドレターキュー

エラー処理のために、必ず DLQ を設定します：

const dlq = new sqs.Queue(this, 'DLQ', {
  retentionPeriod: Duration.days(14),
});

const queue = new sqs.Queue(this, 'Queue', {
  deadLetterQueue: {
    queue: dlq,
    maxReceiveCount: 3,
  },
});

// Monitor DLQ depth
new cloudwatch.Alarm(this, 'DLQAlarm', {
  metric: dlq.metricApproximateNumberOfMessagesVisible(),
  threshold: 1,
  evaluationPeriods: 1,
  alarmDescription: 'Messages in DLQ require attention',
});

可観測性

トレーシングとモニタリングを有効にします：

new NodejsFunction(this, 'Function', {
  entry: 'src/handler.ts',
  tracing: lambda.Tracing.ACTIVE, // X-Ray tracing
  environment: {
    POWERTOOLS_SERVICE_NAME: 'order-service',
    POWERTOOLS_METRICS_NAMESPACE: 'MyApp',
    LOG_LEVEL: 'INFO',
  },
});

MCP サーバーを効果的に活用する

AWS Serverless MCP の使用方法

ライフサイクル管理：

• 新しいサーバーレスプロジェクトを初期化します
• SAM テンプレートを生成します
• アプリケーションをデプロイします
• デプロイ前にローカルでテストします

Lambda Tool MCP の使用方法

関数の実行：

• Lambda 関数を直接テストします
• 自動化ワークフローを実行します
• プライベートリソースにアクセスします
• 統合を検証します

Step Functions MCP の使用方法

ワークフローのオーケストレーション：

• 複雑なワークフロー向けにステートマシンを作成します
• 分散トランザクションを実行します
• Saga パターンを実装します
• マイクロサービスを調整します

SNS/SQS MCP の使用方法

メッセージング操作：

• Pub/Sub パターンをテストします
• キューにテストメッセージを送信します
• イベントルーティングを検証します
• メッセージ処理をデバッグします

追加リソース

このスキルには、AWS のベストプラクティスに基づいた包括的なリファレンスドキュメントが含まれています。

• サーバーレスパターン：references/serverless-patterns.md

  • サーバーレスの中核的なアーキテクチャと API パターン
  • データ処理と統合のパターン
  • Step Functions によるオーケストレーション
  • 回避すべきアンチパターン

• イベント駆動型アーキテクチャのパターン：references/eda-patterns.md

  • イベントのルーティングと処理のパターン
  • イベントソーシングとサーガのパターン
  • 冪等性とエラー処理
  • メッセージの順序付けと重複排除

• セキュリティのベストプラクティス：references/security-best-practices.md

  • 責任共有モデル
  • IAM の最小権限パターン
  • データ保護と暗号化
  • VPC によるネットワークセキュリティ

• オブザーバビリティのベストプラクティス：references/observability-best-practices.md

  • 3 つの柱：メトリクス、ログ、トレース
  • Lambda Powertools による構造化ログ
  • X-Ray による分散トレーシング
  • CloudWatch のアラームとダッシュボード

• パフォーマンスの最適化：references/performance-optimization.md

  • コールドスタートの最適化手法
  • メモリと CPU の最適化
  • パッケージサイズの削減
  • プロビジョニング済み同時実行のパターン

• デプロイのベストプラクティス：references/deployment-best-practices.md

  • CI/CD パイプラインの設計
  • テスト戦略（ユニット、統合、負荷）
  • デプロイ戦略（カナリア、ブルー／グリーン）
  • ロールバックと安全性を確保する仕組み

外部リソース：

• AWS Well-Architected Serverless Lens：https://docs.aws.amazon.com/wellarchitected/latest/serverless-applications-lens/
• ServerlessLand.com：構築済みのサーバーレスパターン
• AWS Serverless Workshops：https://serverlessland.com/learn?type=Workshops

詳細な実装パターン、アンチパターン、コード例については、スキルディレクトリ内の包括的なリファレンスを参照してください。