StepFunctions: 具有 `apigateway:invoke` 的任务在不发出 HTTP 请求的情况下失败

import json import threading import time from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer import boto3 from moto import mock_aws RECEIVED = []

class Handler(BaseHTTPRequestHandler): protocol_version = "HTTP/1.1"

def do_POST(self):
    length = int(self.headers.get("Content-Length") or 0)
    body = self.rfile.read(length) if length else b""
    RECEIVED.append({"path": self.path, "body": body.decode()})
    payload = b'{"ok": true}'
    self.send_response(200)
    self.send_header("Content-Type", "application/json")
    self.send_header("Content-Length", str(len(payload)))
    self.end_headers()
    self.wfile.write(payload)

def log_message(self, *args):
    pass

server = ThreadingHTTPServer(("127.0.0.1", 0), Handler) PORT = server.server_address[1] threading.Thread(target=server.serve_forever, daemon=True).start() DEFN = { "StartAt": "Call", "States": { "Call": { "Type": "Task", "Resource": "arn:AWS:states:::apigateway:invoke", "Parameters": { "ApiEndpoint": f"http://127.0.0.1:{PORT}", "Method": "POST", "Stage": "prod", "Path": "/things", "QueryParameters": {"mode": ["fast"]}, "RequestBody": {"hello": "world"}, …