Architecting Enterprise Model Context Protocol (MCP) Dynamic Tool Execution Timeout Handling & DLQ Trajectory Re-Injection with Amazon Bedrock, AWS Step Functions, and Amazon SQS

import json
import time
import boto3
from typing import Dict, Any, Tuple

class MCPTimeoutAndReinjectionHandler:
    def __init__(self, region_name: str = "us-east-1"):
        self.sfn = boto3.client("stepfunctions", region_name=region_name)
        self.sqs = boto3.client("sqs", region_name=region_name)
        self.dynamodb = boto3.resource("dynamodb", region_name=region_name)
        self.state_table = self.dynamodb.Table("aiconnect_agent_checkpoints")

    def execute_mcp_tool_with_timeout(
        self, task_token: str, session_id: str, tool_name: str, arguments: Dict[str, Any]
    ) -> Tuple[bool, str]:
        print(f"⚡ [MCP Execution Worker] Processing tool '{tool_name}' for session '{session_id}'...")
        start_time = time.time()

        try:
            # Simulate tool processing logic
            if arguments.get("simulate_timeout"):
                raise TimeoutError("Target database connection timed out after 30000ms.")

            result_payload = {
                "status": "SUCCESS",
                "tool": tool_name,
                "output": "Execution completed successfully.",
                "duration_ms": int((time.time() - start_time) * 1000)
            }

            # Callback to Step Functions to resume agent trajectory
            self.sfn.send_task_success(
                taskToken=task_token,
                output=json.dumps(result_payload)
            )
            print(f"✓ [SFN Callback Sent] Task token succeeded for session '{session_id}'.")
            return True, "SUCCESS"

        except Exception as e:
            print(f"🚨 [Tool Execution Failure] Tool '{tool_name}' failed: {str(e)}")
            error_payload = {
                "status": "FAILED",
                "error": str(e),
                "tool": tool_name,
                "session_id": session_id
            }
            # Notify Step Functions of task failure
            try:
                self.sfn.send_task_failure(
                    taskToken=task_token,
                    error="MCPToolExecutionTimeout",
                    cause=json.dumps(error_payload)
                )
            except Exception as sfn_err:
                print(f"⚠️ Could not deliver task failure to SFN: {sfn_err}")

            return False, str(e)

    def process_dlq_reinjection(self, dlq_queue_url: str, target_queue_url: str):
        print(f"🔍 [DLQ Re-Injection Engine] Polling DLQ '{dlq_queue_url}'...")
        response = self.sqs.receive_message(
            QueueUrl=dlq_queue_url,
            MaxNumberOfMessages=5,
            WaitTimeSeconds=5
        )

        messages = response.get("Messages", [])
        if not messages:
            print("ℹ No failed trajectories in DLQ.")
            return

        for msg in messages:
            body = json.loads(msg["Body"])
            session_id = body.get("session_id")
            print(f"🔄 Re-injecting failed trajectory for session '{session_id}' into primary queue...")

            # Re-enqueue to primary task queue
            self.sqs.send_message(
                QueueUrl=target_queue_url,
                MessageBody=json.dumps(body),
                MessageGroupId=session_id
            )

            # Delete from DLQ
            self.sqs.delete_message(
                QueueUrl=dlq_queue_url,
                ReceiptHandle=msg["ReceiptHandle"]
            )
            print(f"✓ Trajectory '{session_id}' successfully re-injected.")

if __name__ == "__main__":
    handler = MCPTimeoutAndReinjectionHandler()
    # Simulated execution test
    success, msg = handler.execute_mcp_tool_with_timeout(
        task_token="sample_token_123",
        session_id="sess_prod_9001",
        tool_name="aws_terraform_apply",
        arguments={"cluster_id": "eks-prod-01"}
    )
    print(f"Result: {success}, Message: {msg}")