Python script to invoke an AWS API Gateway endpoint that requires IAM authentication. Uses boto for signing the request, and requests for issuing it.

apigateway-invoke.py

#!/usr/bin/env python3
"""
Provides a simple Python wrapper for invoking an API Gateway endpoint using IAM signed requests.

Example:
  python3 apigateway-invoke.py GET \
    https://xxxxxxxxxx.execute-api.us-east-1.amazonaws.com/default/MethodName | jq .
"""

try:
    from urllib.parse import urlparse, urlencode, parse_qs
except ImportError:
    from urlparse import urlparse, parse_qs
    from urllib import urlencode

import re
import sys

import requests
from boto3 import Session
from botocore.auth import SigV4Auth
from botocore.awsrequest import AWSRequest


def signing_headers(method, url_string, body):
    # Adapted from:
    #   https://github.com/jmenga/requests-aws-sign/blob/master/requests_aws_sign/requests_aws_sign.py
    region = re.search("execute-api.(.*).amazonaws.com", url_string).group(1)
    url = urlparse(url_string)
    path = url.path or '/'
    querystring = ''
    if url.query:
        querystring = '?' + urlencode(
            parse_qs(url.query, keep_blank_values=True), doseq=True)

    safe_url = url.scheme + '://' + url.netloc.split(
        ':')[0] + path + querystring
    request = AWSRequest(method=method.upper(), url=safe_url, data=body)
    SigV4Auth(Session().get_credentials(), "execute-api",
              region).add_auth(request)
    return dict(request.headers.items())


if __name__ == "__main__":
    method = sys.argv[1]
    url = sys.argv[2]

    if not sys.stdin.isatty():
        body = sys.stdin.read()
    else:
        body = None

    r = requests.get(url, headers=signing_headers(method, url, body))
    print(r.content.decode("utf-8"))

@Riebart - thanks for the response! I needed to transform the body into a JSON object and it worked.

I tried to use json.loads('json:str') with requests.post(url, signed_headers, json=jsonData), it does not work, finally it turned out that if your APIGateWay's auth method is None, you don't need to pass the signed_headers, just do post as:

r = requests.post(url, json)