EventBridge agent backbone: EventBridge as the Agent’s Async Backbone

start_job(args, idempotencyKey)
  → DynamoDB PENDING (conditional)
  → PutEvents JobRequested
  → return {jobId}

Worker rule → run → SUCCEEDED|FAILED + result pointer
get_job(jobId) → status for next agent turn

import uuid, time, json, boto3
ddb = boto3.resource("dynamodb").Table("AgentJobs")
eb = boto3.client("events")

def start_job(tenant: str, tool: str, args: dict, idem: str) -> dict:
    job_id = str(uuid.uuid4())
    try:
        ddb.put_item(
            Item={
                "pk": f"tenant#{tenant}",
                "sk": f"idem#{idem}",
                "jobId": job_id,
                "status": "PENDING",
                "tool": tool,
                "createdAt": int(time.time()),
            },
            ConditionExpression="attribute_not_exists(sk)",
        )
    except ddb.meta.client.exceptions.ConditionalCheckFailedException:
        prev = ddb.get_item(Key={"pk": f"tenant#{tenant}", "sk": f"idem#{idem}"})["Item"]
        return {"jobId": prev["jobId"], "deduped": True}
    eb.put_events(Entries=[{
        "Source": "agent.tools",
        "DetailType": "JobRequested",
        "Detail": json.dumps({"jobId": job_id, "tenant": tenant, "tool": tool, "args": args}),
        "EventBusName": "agents",
    }])
    return {"jobId": job_id, "deduped": False}