Running an aws lambda python openai-compatible api call inside a serverless function introduces cold-start and timeout constraints that don’t exist in long-lived services. This guide walks through packaging a Python Lambda that hits any OpenAI-compatible endpoint, with correct secret handling, retries, and response parsing.
Step 1: Create the execution role and baseline permissions
Lambda needs permission to write logs and read secrets. Create a role with the managed AWSLambdaBasicExecutionRole and add a scoped Secrets Manager read policy. Avoid attaching broad AdministratorAccess—Lambda should only hold what it needs for the call path.
aws iam create-role --role-name lambda-llm-role \
--assume-role-policy-document '{"Version":"2012-10-17","Statement":[{"Effect":"Allow","Principal":{"Service":"lambda.amazonaws.com"},"Action":"sts:AssumeRole"}]}'
aws iam attach-role-policy --role-name lambda-llm-role \
--policy-arn arn:aws:iam::aws:policy/service-role/AWSLambdaBasicExecutionRoleIf your API key lives in Secrets Manager, attach an inline policy restricted to that secret ARN:
{
"Version": "2012-10-17",
"Statement": [{
"Effect": "Allow",
"Action": "secretsmanager:GetSecretValue",
"Resource": "arn:aws:secretsmanager:us-east-1:123456789012:secret:llm-api-key-ABCD"
}]
}Keep the function out of a VPC unless you must reach private resources. A public HTTPS call to an OpenAI-compatible endpoint works from the Lambda default network and avoids ENI cold-start penalties.
Step 2: Choose a packaging strategy
For a dependency on httpx, a container image gives reproducible builds and avoids wheel compatibility issues. The aws lambda python openai-compatible api client is small, but the container base handles the interpreter and pip layers cleanly.
FROM public.ecr.aws/lambda/python:3.12
COPY requirements.txt ${LAMBDA_TASK_ROOT}
RUN pip install --no-cache-dir -r requirements.txt
COPY app.py ${LAMBDA_TASK_ROOT}
CMD ["app.handler"]requirements.txt should be minimal:
httpx==0.27.0
boto3==1.34.0Zip packaging works if you use a Lambda layer with prebuilt wheels, but you then manage two artifacts. Containers collapse that into one image. Build time increases slightly, but local testing with docker run matches the runtime exactly.
Step 3: Implement the handler
The handler opens an async HTTP session, posts to /v1/chat/completions, and returns the message content. Strip the official SDK—its weight hurts cold start and you only need one POST shape.
import os
import httpx
import json
import boto3
from botocore.exceptions import ClientError
BASE_URL = os.environ["BASE_URL"]
MODEL = os.environ.get("MODEL", "gpt-3.5-turbo")
_secret_cache = {}
def get_secret(secret_id):
if secret_id in _secret_cache:
return _secret_cache[secret_id]
client = boto3.client("secretsmanager")
val = client.get_secret_value(SecretId=secret_id)["SecretString"]
_secret_cache[secret_id] = val
return val
async def handler(event, context):
try:
api_key = get_secret(os.environ["API_KEY_SECRET"])
messages = event.get("messages", [{"role": "user", "content": "Say hi."}])
async with httpx.AsyncClient(timeout=25.0) as client:
resp = await client.post(
f"{BASE_URL}/v1/chat/completions",
headers={"Authorization": f"Bearer {api_key}", "Content-Type": "application/json"},
json={"model": MODEL, "messages": messages, "max_tokens": 200},
)
resp.raise_for_status()
data = resp.json()
return {
"statusCode": 200,
"body": json.dumps({"content": data["choices"][0]["message"]["content"]}),
}
except ClientError as e:
return {"statusCode": 500, "body": json.dumps({"error": "secret_load_failed"})}
except httpx.HTTPStatusError as e:
return {"statusCode": e.response.status_code, "body": json.dumps({"error": "upstream_error"})}The timeout is set to 25 seconds, below the Lambda max, to leave headroom for serialization. If you front this with API Gateway, the statusCode and body map directly to proxy responses.
Step 4: Externalize configuration and secrets
Environment variables in Lambda are limited to 4 KB total. Put non-secret config (BASE_URL, MODEL) in env vars; put the key in Secrets Manager. Cache the secret in a module-level dict so subsequent warm invokes skip the boto3 round trip.
# already shown in handler; set env vars via deploy config
# BASE_URL=https://api.openai.com
# MODEL=gpt-4o-mini
# API_KEY_SECRET=llm-api-keyIf you point BASE_URL at a gateway such as n4n.ai, which exposes one OpenAI-compatible endpoint across 240+ models with automatic fallback when a provider is rate-limited, you can keep MODEL as a routing hint and drop custom multi-vendor logic from the Lambda entirely.
Step 5: Retry, fallback, and degradation
Providers return 429 or 5xx under load. Implement a bounded retry with exponential backoff. For an aws lambda python openai-compatible api client, this is the difference between sporadic failures and a stable endpoint.
from asyncio import sleep
async def post_with_retry(client, url, headers, payload, attempts=3):
last_exc = None
for i in range(attempts):
try:
resp = await client.post(url, headers=headers, json=payload)
if resp.status_code >= 500 or resp.status_code == 429:
await sleep(2 ** i)
continue
resp.raise_for_status()
return resp
except httpx.RequestError as e:
last_exc = e
await sleep(2 ** i)
raise last_exc or RuntimeError("retry_exhausted")Wire post_with_retry into the handler instead of the raw client.post. In production, prefer tenacity with @retry(wait=wait_exponential(), stop=stop_after_attempt(3)) for cleaner code. If your gateway already handles degraded backends, you can reduce local attempts to one and let it route.
Step 6: Deploy with infrastructure as code
CLI works for a quick test, but Terraform keeps the config reproducible.
resource "aws_lambda_function" "llm" {
function_name = "llm-proxy"
package_type = "Image"
image_uri = "${aws_ecr_repository.llm.repository_url}:latest"
role = aws_iam_role.lambda_llm.arn
timeout = 30
memory_size = 256
environment {
variables = {
BASE_URL = "https://api.openai.com"
MODEL = "gpt-4o-mini"
API_KEY_SECRET = "llm-api-key"
}
}
}Set memory to 256 MB for small payloads; raise to 512 MB if you parse large completions. Container cold starts add 1–3 seconds of latency on first invoke—schedule a CloudWatch Events ping every 5 minutes if you need consistent responsiveness.
Step 7: Verify end-to-end
Invoke from the CLI with a minimal event:
aws lambda invoke \
--function-name llm-proxy \
--payload '{"messages":[{"role":"user","content":"What is 2+2?"}]}' \
response.json
cat response.jsonA successful run returns {"content": "2 + 2 equals 4."} (or equivalent). Check CloudWatch Logs for the request ID and any secret_load_failed entries. If you see Task timed out, lower max_tokens or raise the Lambda timeout. For API Gateway fronting, curl the endpoint with the same JSON body to confirm the proxy integration.
Troubleshooting and operational notes
Cold starts: Container images incur longer init than zip bundles. Keep the image small—no compiler toolchain after pip install.
Payload limits: Lambda request/response via Invoke is capped at 6 MB. Streaming requires Lambda URLs or API Gateway WebSocket; the synchronous Invoke buffers the full body.
Concurrency: Set reserved concurrency to protect downstream provider rate limits. Unbounded scaling will trip 429s faster than your retry can absorb.
Idempotency: If a client retries a Lambda invoke, the upstream POST may duplicate. Pass an Idempotency-Key header if your endpoint supports it, or accept at-most-once semantics for non-critical generations.
The aws lambda python openai-compatible api integration is stable once you separate config, secrets, and retry policy. Treat the function as a thin transport, not a place to embed model-selection business logic.