LLM-Powered SEC Filing Pipeline
Deploy a distributed LLM pipeline that analyzes SEC filings on AWS.
Introduction
30 Second Summary
Every quarter, thousands of public companies file financial reports with the SEC. Analysts spend hours reading through these documents to extract insights about risk, revenue, and strategy.
In this project, you will build a distributed pipeline on AWS that automatically fetches real SEC filings, sends them to an AI model for analysis, and serves structured financial insights through a live public API.
What You'll Build
A distributed SEC filing analysis system using AWS Lambda, SQS, DynamoDB, and OpenAI GPT-4o-mini to process real financial documents and return structured insights via a REST API.
By the end of this project, you'll have:
- A live public API URL with interactive Swagger documentation that anyone can use to trigger a financial analysis.
- A distributed event-driven pipeline using SQS for async messaging between two Lambda microservices.
- An LLM-powered analysis integration that processes real SEC 10-K filings with GPT-4o-mini and stores structured insights in DynamoDB.
- Secret Mission: Add a comparison endpoint that analyzes two companies side-by-side.
Are there any prerequisites?
You'll need an AWS account, VS Code, Python experience, and an OpenAI API key with credits. You'll install the AWS CLI and AWS SAM CLI during Step 1.
Build the FastAPI Ingestion Service
Your environment is locked and loaded. Now it is time to build the core service that powers your pipeline.
In this step, you will create the FastAPI application that accepts a company ticker symbol, fetches real SEC filing data, and queues a processing job for analysis. This service is the entry point for your entire distributed pipeline.
In this step, get ready to:
- Create the FastAPI application with endpoints for submitting analysis requests and retrieving results.
- Implement the SEC EDGAR API client to fetch company filing metadata.
- Add the Mangum adapter to make FastAPI compatible with Lambda.
Create the FastAPI application with endpoints
Your API needs three endpoints: one to accept ticker symbols and kick off analysis, one to retrieve completed results, and a health check. Before writing those endpoints, you need to set up the app skeleton with its AWS service connections.
- In VS Code, open the file api/app.py inside your sec-analysis-pipeline project folder.
- Add the following imports and app configuration:
import os
import json
import uuid
from datetime import datetime
import boto3
import httpx
from fastapi import FastAPI, HTTPException
from mangum import Mangum
# Initialize the FastAPI application with metadata for Swagger docs
app = FastAPI(
title="SEC Filing Analysis Pipeline",
description="LLM-powered analysis of SEC 10-K filings via a distributed pipeline",
version="1.0.0",
)
# Create AWS service clients for SQS messaging and DynamoDB storage
sqs = boto3.client("sqs")
dynamodb = boto3.resource("dynamodb")
# Read resource identifiers from environment variables (set by SAM template)
QUEUE_URL = os.environ["ANALYSIS_QUEUE_URL"]
TABLE_NAME = os.environ["RESULTS_TABLE_NAME"]
table = dynamodb.Table(TABLE_NAME)
# SEC EDGAR API configuration
SEC_EDGAR_BASE = "https://data.sec.gov"
SEC_HEADERS = {"User-Agent": "SECAnalysisPipeline/1.0 (contact@example.com)"}What does this code do?
This block sets up the foundation for your entire service.
- boto3 connects to Amazon SQS (for queuing jobs) and Amazon DynamoDB (for storing results).
- QUEUE_URL and TABLE_NAME are read from environment variables. Your SAM template will inject these at deploy time.
- SEC_HEADERS includes a User-Agent string because the SEC EDGAR API requires every request to identify the caller.
Now add the endpoint that accepts a ticker symbol and kicks off the analysis pipeline. Add this code below the SEC_HEADERS line:
- Add the POST endpoint for submitting analysis requests:
@app.post("/analyze/{ticker}")
def submit_analysis(ticker: str):
"""Submit a company ticker for SEC filing analysis."""
# Look up the company and fetch its latest 10-K filing
cik = get_company_cik(ticker)
filing_text = get_latest_10k_text(cik)
# Generate a unique ID to track this analysis
analysis_id = str(uuid.uuid4())
# Store a placeholder record in DynamoDB with "processing" status
table.put_item(
Item={
"analysis_id": analysis_id,
"ticker": ticker.upper(),
"status": "processing",
"submitted_at": datetime.utcnow().isoformat(),
}
)
# Send the filing text to SQS for the worker Lambda to pick up
sqs.send_message(
QueueUrl=QUEUE_URL,
MessageBody=json.dumps(
{
"analysis_id": analysis_id,
"ticker": ticker.upper(),
"cik": cik,
"filing_text": filing_text,
}
),
)
return {
"analysis_id": analysis_id,
"ticker": ticker.upper(),
"status": "processing",
"message": "Analysis submitted. Use GET /results/{analysis_id} to retrieve results.",
}What does this endpoint do?
This is the heart of your ingestion service. When someone POSTs a ticker symbol like AAPL, the function:
- Fetches the company's SEC filing (using helper functions you will write next).
- Creates a tracking record in DynamoDB so the caller can check status later.
- Sends the filing text to an SQS queue, where a separate worker Lambda will pick it up for LLM analysis.
This pattern decouples the slow LLM processing from the fast API response. The caller gets an immediate acknowledgment with an analysis_id to poll later.
- Add the GET endpoints for retrieving results and checking service health:
@app.get("/results/{analysis_id}")
def get_results(analysis_id: str):
"""Retrieve analysis results by ID."""
response = table.get_item(Key={"analysis_id": analysis_id})
item = response.get("Item")
if not item:
raise HTTPException(status_code=404, detail="Analysis not found")
return item
@app.get("/health")
def health():
"""Health check endpoint."""
return {"status": "healthy", "service": "sec-analysis-api"}What do these endpoints do?
- /results/{analysis_id} looks up a completed analysis in DynamoDB and returns it. If the analysis is still processing, the status field will say processing.
- /health is a simple endpoint that confirms the service is running. Load balancers and monitoring tools use this to verify availability.
- Save api/app.py.
Your file now has the app skeleton and all three endpoints. The submit_analysis function references two helper functions that do not exist yet. You will write those next.
Implement the SEC EDGAR API client
The SEC's EDGAR database holds every public company's financial filings. Your app needs two helper functions: one to look up a company's internal SEC identifier (called a CIK number), and one to fetch the actual 10-K filing text.
What is a 10-K filing?
A 10-K is the annual report that every publicly traded company must file with the SEC. It contains detailed financial statements, risk factors, and business descriptions. This is the raw data your LLM will analyze.
- In api/app.py, find the line that says SEC_HEADERS = {"User-Agent": "SECAnalysisPipeline/1.0 (contact@example.com)"}.
- Add the following function below it (above the @app.post decorator):
def get_company_cik(ticker: str) -> str:
"""Look up a company's CIK number from its ticker symbol."""
# Fetch the SEC's master list of all company tickers
response = httpx.get(
"https://www.sec.gov/files/company_tickers.json",
headers=SEC_HEADERS,
)
response.raise_for_status()
tickers_data = response.json()
# Search for the matching ticker and return its CIK padded to 10 digits
for entry in tickers_data.values():
if entry["ticker"].upper() == ticker.upper():
return str(entry["cik_str"]).zfill(10)
raise HTTPException(status_code=404, detail=f"Ticker '{ticker}' not found in SEC database")What does this function do?
Every company in the SEC system has a unique CIK (Central Index Key) number. This function downloads the SEC's public ticker-to-CIK mapping and finds the matching company.
The CIK is zero-padded to 10 digits because that is the format the EDGAR submissions API expects.
- Add the second helper function directly below get_company_cik (still above @app.post):
def get_latest_10k_text(cik: str) -> str:
"""Fetch the most recent 10-K filing description from SEC EDGAR."""
submissions_url = f"{SEC_EDGAR_BASE}/submissions/CIK{cik}.json"
response = httpx.get(submissions_url, headers=SEC_HEADERS)
response.raise_for_status()
filings = response.json()
# Search through recent filings for the first 10-K
recent = filings["filings"]["recent"]
for i, form in enumerate(recent["form"]):
if form == "10-K":
# Build the URL to the actual filing document
accession = recent["accessionNumber"][i].replace("-", "")
primary_doc = recent["primaryDocument"][i]
filing_url = f"https://www.sec.gov/Archives/edgar/data/{cik.lstrip('0')}/{accession}/{primary_doc}"
filing_response = httpx.get(filing_url, headers=SEC_HEADERS, timeout=30.0)
filing_response.raise_for_status()
# Truncate to 15000 chars to stay within LLM context limits
text = filing_response.text[:15000]
return text
raise HTTPException(status_code=404, detail="No 10-K filing found for this company")What does this function do?
This function does the heavy lifting of fetching actual SEC data. It queries the EDGAR submissions API to find the company's most recent 10-K annual filing, then downloads the filing document itself.
The text is truncated to 15,000 characters. This keeps it within GPT-4o-mini's context window while still capturing the most important sections of the filing.
- Save api/app.py.
Your app can now resolve any public company ticker to its SEC filing data. The submit_analysis endpoint's calls to these functions are now complete.
Confused about the file order?
Make sure both get_company_cik and get_latest_10k_text are placed between the SEC_HEADERS constant and the @app.post("/analyze/{ticker}") decorator. Python needs functions defined before they are called.
help me figure out the correct order of functions in my app.py file
Add the Mangum adapter and dependencies
AWS Lambda does not speak the same protocol as FastAPI. Lambda sends raw event dictionaries, but FastAPI expects ASGI requests. Mangum is an adapter that translates between these two formats.
- At the very bottom of api/app.py (after the health function), add this line:
handler = Mangum(app)Why is this one line so important?
This single line is what makes your entire FastAPI app deployable on AWS Lambda. When Lambda receives a request through API Gateway, it calls this handler function. Mangum converts the Lambda event into an ASGI request, passes it to FastAPI, and converts the response back into the format Lambda expects.
- Save api/app.py.
- Now create the dependencies file. Open api/requirements.txt and add the following:
fastapi==0.115.6
mangum==0.19.0
httpx==0.28.1
boto3==1.35.86What are these packages?
- fastapi is the web framework for building your API endpoints.
- mangum is the Lambda-to-ASGI adapter you just wired up.
- httpx is an HTTP client for calling the SEC EDGAR API.
- boto3 is the AWS SDK for Python, used to interact with SQS and DynamoDB.
- Save api/requirements.txt.
- Confirm both files exist by running:
ls api/You should see app.py and requirements.txt in the output.
Don't see both files?
Make sure you created the files inside the api/ subdirectory, not in the project root. If they ended up in the wrong place, move them with:
mv app.py api/
mv requirements.txt api/help me check that my project directory structure is correct
✔️ Awesome, I've got everything!
Great. Double check you have saved both files.
ⓧ I'd like to double check the full code
Here is the complete api/app.py file:
import os
import json
import uuid
from datetime import datetime
import boto3
import httpx
from fastapi import FastAPI, HTTPException
from mangum import Mangum
app = FastAPI(
title="SEC Filing Analysis Pipeline",
description="LLM-powered analysis of SEC 10-K filings via a distributed pipeline",
version="1.0.0",
)
sqs = boto3.client("sqs")
dynamodb = boto3.resource("dynamodb")
QUEUE_URL = os.environ["ANALYSIS_QUEUE_URL"]
TABLE_NAME = os.environ["RESULTS_TABLE_NAME"]
table = dynamodb.Table(TABLE_NAME)
SEC_EDGAR_BASE = "https://data.sec.gov"
SEC_HEADERS = {"User-Agent": "SECAnalysisPipeline/1.0 (contact@example.com)"}
def get_company_cik(ticker: str) -> str:
"""Look up a company's CIK number from its ticker symbol."""
response = httpx.get(
"https://www.sec.gov/files/company_tickers.json",
headers=SEC_HEADERS,
)
response.raise_for_status()
tickers_data = response.json()
for entry in tickers_data.values():
if entry["ticker"].upper() == ticker.upper():
return str(entry["cik_str"]).zfill(10)
raise HTTPException(status_code=404, detail=f"Ticker '{ticker}' not found in SEC database")
def get_latest_10k_text(cik: str) -> str:
"""Fetch the most recent 10-K filing description from SEC EDGAR."""
submissions_url = f"{SEC_EDGAR_BASE}/submissions/CIK{cik}.json"
response = httpx.get(submissions_url, headers=SEC_HEADERS)
response.raise_for_status()
filings = response.json()
recent = filings["filings"]["recent"]
for i, form in enumerate(recent["form"]):
if form == "10-K":
accession = recent["accessionNumber"][i].replace("-", "")
primary_doc = recent["primaryDocument"][i]
filing_url = f"https://www.sec.gov/Archives/edgar/data/{cik.lstrip('0')}/{accession}/{primary_doc}"
filing_response = httpx.get(filing_url, headers=SEC_HEADERS, timeout=30.0)
filing_response.raise_for_status()
text = filing_response.text[:15000]
return text
raise HTTPException(status_code=404, detail="No 10-K filing found for this company")
@app.post("/analyze/{ticker}")
def submit_analysis(ticker: str):
"""Submit a company ticker for SEC filing analysis."""
cik = get_company_cik(ticker)
filing_text = get_latest_10k_text(cik)
analysis_id = str(uuid.uuid4())
table.put_item(
Item={
"analysis_id": analysis_id,
"ticker": ticker.upper(),
"status": "processing",
"submitted_at": datetime.utcnow().isoformat(),
}
)
sqs.send_message(
QueueUrl=QUEUE_URL,
MessageBody=json.dumps(
{
"analysis_id": analysis_id,
"ticker": ticker.upper(),
"cik": cik,
"filing_text": filing_text,
}
),
)
return {
"analysis_id": analysis_id,
"ticker": ticker.upper(),
"status": "processing",
"message": "Analysis submitted. Use GET /results/{analysis_id} to retrieve results.",
}
@app.get("/results/{analysis_id}")
def get_results(analysis_id: str):
"""Retrieve analysis results by ID."""
response = table.get_item(Key={"analysis_id": analysis_id})
item = response.get("Item")
if not item:
raise HTTPException(status_code=404, detail="Analysis not found")
return item
@app.get("/health")
def health():
"""Health check endpoint."""
return {"status": "healthy", "service": "sec-analysis-api"}
handler = Mangum(app)And here is the complete api/requirements.txt:
fastapi==0.115.6
mangum==0.19.0
httpx==0.28.1
boto3==1.35.86Your ingestion service is complete. It can resolve any public company ticker to real SEC data and queue it for LLM analysis. Next up, you will build the worker Lambda that picks jobs off the queue and sends them to GPT-4o-mini.
Build the SQS Worker Lambda
Your API service is ready to accept tickers and drop messages onto an SQS queue. But nothing is listening on the other end yet.
In this step, you'll build the worker Lambda that picks up those messages, sends the filing text to OpenAI GPT-4o-mini for analysis, and writes structured results back to DynamoDB.
In this step, get ready to:
- Create the SQS worker function with DynamoDB and OpenAI clients.
- Define the analysis prompt that instructs GPT-4o-mini to return structured JSON.
- Build the message handler that processes filings and stores results.
Create the worker module with its clients
The worker Lambda runs as a completely separate microservice from your API. It has its own imports, its own boto3 clients, and its own entry point that SQS triggers automatically whenever a new message arrives.
- Open worker/handler.py in VS Code (this file should be empty) and add the following imports and client setup:
import os
import json
from datetime import datetime
import boto3
from openai import OpenAI
dynamodb = boto3.resource("dynamodb")
TABLE_NAME = os.environ["RESULTS_TABLE_NAME"]
table = dynamodb.Table(TABLE_NAME)
openai_client = OpenAI(api_key=os.environ["OPENAI_API_KEY"])What does this code do?
This module-level setup runs once when the Lambda container starts. It creates a DynamoDB table reference and an OpenAI client, both of which persist across invocations for performance.
- RESULTS_TABLE_NAME and OPENAI_API_KEY are environment variables that SAM will inject at deploy time.
- The openai_client is initialized outside the handler so the connection pool is reused across SQS messages.
You won't see a change yet. The next section defines the prompt this worker sends to GPT-4o-mini.
Define the analysis prompt for GPT-4o-mini
The quality of your LLM output depends entirely on the prompt. You need GPT-4o-mini to return a consistent JSON structure so you can store the analysis reliably in DynamoDB.
- Below the openai_client line, add the prompt template:
ANALYSIS_PROMPT = """You are a financial analyst. Analyze the following SEC 10-K filing excerpt and provide a structured analysis in JSON format with these fields:
- company_name: The company's full legal name
- fiscal_year: The fiscal year of the filing
- revenue_summary: A 2-3 sentence summary of revenue and financial performance
- risk_factors: A list of the top 3 risk factors mentioned
- business_summary: A 2-3 sentence summary of what the company does
- key_metrics: Any specific financial numbers mentioned (revenue, net income, etc.)
Respond ONLY with valid JSON. No markdown, no code fences.
Filing text:
{filing_text}"""What does this prompt do?
The prompt defines the exact JSON schema you expect back from GPT-4o-mini. By listing each field and its format, you make the model's output predictable enough to parse with json.loads().
The {filing_text} placeholder gets filled at runtime with the SEC filing content pulled from the SQS message.
Why GPT-4o-mini instead of GPT-4o?
GPT-4o-mini costs $0.15 per 1M input tokens and $0.60 per 1M output tokens. A full GPT-4o call would cost 10x more for this structured extraction task. The mini model is fast, cheap, and accurate enough for pulling key fields from financial text.
You won't see a change yet. The next section builds the handler function that calls this prompt.
Build the message handler with error recovery
The handler function is the entry point that SQS invokes. It loops through each message in the batch, calls OpenAI, parses the JSON response, and updates DynamoDB with results or an error status.
- Below ANALYSIS_PROMPT, add the handler function:
def handler(event, context):
"""Process SQS messages containing SEC filing text for LLM analysis."""
for record in event["Records"]:
body = json.loads(record["body"])
analysis_id = body["analysis_id"]
ticker = body["ticker"]
filing_text = body["filing_text"]
try:
completion = openai_client.chat.completions.create(
model="gpt-4o-mini",
messages=[
{
"role": "user",
"content": ANALYSIS_PROMPT.format(filing_text=filing_text),
}
],
temperature=0.2,
)
analysis_text = completion.choices[0].message.content
analysis_data = json.loads(analysis_text)
table.update_item(
Key={"analysis_id": analysis_id},
UpdateExpression="SET #s = :status, analysis = :analysis, completed_at = :completed",
ExpressionAttributeNames={"#s": "status"},
ExpressionAttributeValues={
":status": "completed",
":analysis": analysis_data,
":completed": datetime.utcnow().isoformat(),
},
)
except Exception as e:
table.update_item(
Key={"analysis_id": analysis_id},
UpdateExpression="SET #s = :status, error_message = :error",
ExpressionAttributeNames={"#s": "status"},
ExpressionAttributeValues={
":status": "failed",
":error": str(e),
},
)What does this handler do?
- event["Records"] contains the batch of SQS messages. Each record's body holds the JSON your API Lambda sent (analysis_id, ticker, filing_text).
- openai_client.chat.completions.create() sends the filing text to GPT-4o-mini with a low temperature (0.2) for consistent, factual output.
- On success, the DynamoDB item gets updated from processing to completed with the parsed analysis data.
- On failure (bad JSON, API timeout, quota exceeded), the item gets marked failed with the error message. This prevents silent data loss.
Why use update_item instead of put_item?
Your API Lambda already created the DynamoDB record with a processing status. The worker updates that existing record rather than overwriting it, preserving the original submitted_at timestamp and ticker.
- Now create the worker's dependency file. Open worker/requirements.txt and add:
openai==1.58.1
boto3==1.35.86Why does the worker have its own requirements.txt?
Each Lambda function gets deployed as a separate package. The worker only needs the OpenAI SDK and boto3. Keeping dependencies minimal reduces cold start time and package size.
- Save both worker/handler.py and worker/requirements.txt.
- Confirm both files exist by running:
ls worker/- You should see handler.py and requirements.txt in the output.
Only seeing one file or none?
- Make sure you're in the sec-analysis-pipeline root directory, not inside the worker/ folder itself.
- Check that you saved both files in the worker/ subdirectory, not the project root.
- help me find my worker files
✔️ Awesome, I've got everything!
Great. Make sure both files are saved before moving on.
ⓧ I'd like to double check the full code
Here is the complete worker/handler.py:
import os
import json
from datetime import datetime
import boto3
from openai import OpenAI
dynamodb = boto3.resource("dynamodb")
TABLE_NAME = os.environ["RESULTS_TABLE_NAME"]
table = dynamodb.Table(TABLE_NAME)
openai_client = OpenAI(api_key=os.environ["OPENAI_API_KEY"])
ANALYSIS_PROMPT = """You are a financial analyst. Analyze the following SEC 10-K filing excerpt and provide a structured analysis in JSON format with these fields:
- company_name: The company's full legal name
- fiscal_year: The fiscal year of the filing
- revenue_summary: A 2-3 sentence summary of revenue and financial performance
- risk_factors: A list of the top 3 risk factors mentioned
- business_summary: A 2-3 sentence summary of what the company does
- key_metrics: Any specific financial numbers mentioned (revenue, net income, etc.)
Respond ONLY with valid JSON. No markdown, no code fences.
Filing text:
{filing_text}"""
def handler(event, context):
"""Process SQS messages containing SEC filing text for LLM analysis."""
for record in event["Records"]:
body = json.loads(record["body"])
analysis_id = body["analysis_id"]
ticker = body["ticker"]
filing_text = body["filing_text"]
try:
completion = openai_client.chat.completions.create(
model="gpt-4o-mini",
messages=[
{
"role": "user",
"content": ANALYSIS_PROMPT.format(filing_text=filing_text),
}
],
temperature=0.2,
)
analysis_text = completion.choices[0].message.content
analysis_data = json.loads(analysis_text)
table.update_item(
Key={"analysis_id": analysis_id},
UpdateExpression="SET #s = :status, analysis = :analysis, completed_at = :completed",
ExpressionAttributeNames={"#s": "status"},
ExpressionAttributeValues={
":status": "completed",
":analysis": analysis_data,
":completed": datetime.utcnow().isoformat(),
},
)
except Exception as e:
table.update_item(
Key={"analysis_id": analysis_id},
UpdateExpression="SET #s = :status, error_message = :error",
ExpressionAttributeNames={"#s": "status"},
ExpressionAttributeValues={
":status": "failed",
":error": str(e),
},
)And the complete worker/requirements.txt:
openai==1.58.1
boto3==1.35.86Your two microservices are complete. The API Lambda handles ingestion and retrieval. The worker Lambda handles LLM analysis and storage. Next up, you'll wire them together with a SAM template that defines the full AWS infrastructure.
Define the Infrastructure with SAM
Your application code is ready, but AWS does not know how to deploy it yet. You need to describe your infrastructure in a way that AWS can read, create, and manage automatically.
AWS SAM uses a single YAML template to define everything your pipeline needs. In this step, you will declare your SQS queue, DynamoDB table, both Lambda functions, and their permissions in one file.
In this step, get ready to:
- Write the SAM template declaring all AWS resources for the pipeline.
- Configure least-privilege IAM permissions for each Lambda function.
- Create the deployment configuration file with your OpenAI API key.
Create the SAM template with core resources
Infrastructure-as-code means every AWS resource your pipeline uses is declared in a template file. SAM reads this template and creates the resources for you.
- Create a new file called template.yaml in the root of your sec-analysis-pipeline folder.
- Add the template header, global defaults, parameters, and data layer resources by pasting this code:
AWSTemplateFormatVersion: "2010-09-09"
Transform: AWS::Serverless-2016-10-31
Description: SEC Filing Analysis Pipeline - LLM-powered distributed analysis
Globals:
Function:
Timeout: 60
Runtime: python3.12
MemorySize: 256
Parameters:
OpenAIApiKey:
Type: String
NoEcho: true
Description: OpenAI API key for GPT-4o-mini
Resources:
AnalysisQueue:
Type: AWS::SQS::Queue
Properties:
QueueName: sec-analysis-queue
VisibilityTimeout: 120
ResultsTable:
Type: AWS::DynamoDB::Table
Properties:
TableName: sec-analysis-results
BillingMode: PAY_PER_REQUEST
AttributeDefinitions:
- AttributeName: analysis_id
AttributeType: S
KeySchema:
- AttributeName: analysis_id
KeyType: HASHWhat does this code do?
- Globals sets defaults for all Lambda functions: a 60-second timeout, Python 3.12 runtime, and 256 MB of memory.
- Parameters defines your OpenAI API key as a secure input. The NoEcho: true setting hides it in logs and the CloudFormation console.
- AnalysisQueue creates an SQS queue with a 120-second visibility timeout. This gives the worker Lambda enough time to finish processing before the message becomes visible again.
- ResultsTable creates a DynamoDB table with on-demand billing (pay per request, no provisioned capacity). The partition key is analysis_id.
You won't see a change yet. The next chunk completes the template with your Lambda definitions.
Add the Lambda functions with IAM policies
Each Lambda function needs to know which code to run, which resources it can access, and what triggers it. The Policies section enforces least-privilege access, so each function can only reach the resources it needs.
- In template.yaml, add the following below the ResultsTable resource you just defined:
ApiFunction:
Type: AWS::Serverless::Function
Properties:
FunctionName: sec-analysis-api
Handler: app.handler
CodeUri: api/
Environment:
Variables:
ANALYSIS_QUEUE_URL: !Ref AnalysisQueue
RESULTS_TABLE_NAME: !Ref ResultsTable
Policies:
- SQSSendMessagePolicy:
QueueName: !GetAtt AnalysisQueue.QueueName
- DynamoDBCrudPolicy:
TableName: !Ref ResultsTable
Events:
Api:
Type: HttpApi
WorkerFunction:
Type: AWS::Serverless::Function
Properties:
FunctionName: sec-analysis-worker
Handler: handler.handler
CodeUri: worker/
Timeout: 120
MemorySize: 512
Environment:
Variables:
RESULTS_TABLE_NAME: !Ref ResultsTable
OPENAI_API_KEY: !Ref OpenAIApiKey
Policies:
- DynamoDBCrudPolicy:
TableName: !Ref ResultsTable
Events:
SQSEvent:
Type: SQS
Properties:
Queue: !GetAtt AnalysisQueue.Arn
BatchSize: 1
Outputs:
ApiUrl:
Description: Public API URL
Value: !Sub "https://${ServerlessHttpApi}.execute-api.${AWS::Region}.amazonaws.com"What does this code do?
- ApiFunction points to your FastAPI code in api/. The HttpApi event creates an API Gateway HTTP API endpoint automatically.
- WorkerFunction points to worker/ and gets triggered by SQS messages. The BatchSize: 1 setting processes one filing at a time.
- Environment variables inject resource references at runtime. !Ref AnalysisQueue resolves to the queue URL. !Ref OpenAIApiKey resolves to the parameter value you pass at deploy time.
- Outputs prints your public API URL after deployment completes.
How do the IAM policies enforce least privilege?
Each Lambda gets only the permissions it needs. The API function can send messages to SQS and read/write DynamoDB, but it has no access to your OpenAI API key.
The worker function can read/write DynamoDB and receives the OpenAI key, but it cannot send SQS messages. If the API function were ever compromised, your OpenAI credentials would remain safe.
- Save template.yaml.
- Validate the template syntax by running:
sam validateYou should see a message confirming the template is valid.
Seeing a validation error?
- Check that your YAML indentation uses spaces, not tabs. YAML is strict about whitespace.
- Make sure ApiFunction and WorkerFunction are at the same indentation level as AnalysisQueue and ResultsTable.
- Verify that Outputs is at the same level as Resources (no extra indentation).
help me fix a SAM template validation error
✔️ Awesome, I've got everything!
Great. Make sure you have saved template.yaml and sam validate passes.
ⓧ I'd like to double check the full code
AWSTemplateFormatVersion: "2010-09-09"
Transform: AWS::Serverless-2016-10-31
Description: SEC Filing Analysis Pipeline - LLM-powered distributed analysis
Globals:
Function:
Timeout: 60
Runtime: python3.12
MemorySize: 256
Parameters:
OpenAIApiKey:
Type: String
NoEcho: true
Description: OpenAI API key for GPT-4o-mini
Resources:
AnalysisQueue:
Type: AWS::SQS::Queue
Properties:
QueueName: sec-analysis-queue
VisibilityTimeout: 120
ResultsTable:
Type: AWS::DynamoDB::Table
Properties:
TableName: sec-analysis-results
BillingMode: PAY_PER_REQUEST
AttributeDefinitions:
- AttributeName: analysis_id
AttributeType: S
KeySchema:
- AttributeName: analysis_id
KeyType: HASH
ApiFunction:
Type: AWS::Serverless::Function
Properties:
FunctionName: sec-analysis-api
Handler: app.handler
CodeUri: api/
Environment:
Variables:
ANALYSIS_QUEUE_URL: !Ref AnalysisQueue
RESULTS_TABLE_NAME: !Ref ResultsTable
Policies:
- SQSSendMessagePolicy:
QueueName: !GetAtt AnalysisQueue.QueueName
- DynamoDBCrudPolicy:
TableName: !Ref ResultsTable
Events:
Api:
Type: HttpApi
WorkerFunction:
Type: AWS::Serverless::Function
Properties:
FunctionName: sec-analysis-worker
Handler: handler.handler
CodeUri: worker/
Timeout: 120
MemorySize: 512
Environment:
Variables:
RESULTS_TABLE_NAME: !Ref ResultsTable
OPENAI_API_KEY: !Ref OpenAIApiKey
Policies:
- DynamoDBCrudPolicy:
TableName: !Ref ResultsTable
Events:
SQSEvent:
Type: SQS
Properties:
Queue: !GetAtt AnalysisQueue.Arn
BatchSize: 1
Outputs:
ApiUrl:
Description: Public API URL
Value: !Sub "https://${ServerlessHttpApi}.execute-api.${AWS::Region}.amazonaws.com"Create the deployment configuration
The SAM template declares what resources to create, but it does not know your specific settings like the AWS region or your API key. A deployment configuration file stores these values so you do not have to type them every time you deploy.
- Create a new file called samconfig.toml in the root of your sec-analysis-pipeline folder.
- Paste the following deployment configuration:
version = 0.1
[default.deploy.parameters]
stack_name = "sec-analysis-pipeline"
resolve_s3 = true
s3_prefix = "sec-analysis-pipeline"
region = "us-east-1"
confirm_changeset = true
capabilities = "CAPABILITY_IAM"
parameter_overrides = "OpenAIApiKey=your-api-key-here"What does this configuration do?
- stack_name names your CloudFormation stack. All resources are grouped under this name.
- resolve_s3 = true lets SAM auto-create an S3 bucket for deployment artifacts (your zipped code).
- capabilities = "CAPABILITY_IAM" grants SAM permission to create IAM roles for your Lambda functions.
- parameter_overrides passes your OpenAI API key to the template's OpenAIApiKey parameter. You will replace your-api-key-here with your real key before deploying.
Keep your API key out of version control
If you plan to push this project to GitHub, add samconfig.toml to your .gitignore file. This file contains your OpenAI API key in plain text.
- Save samconfig.toml.
- Confirm both new files exist by running:
ls template.yaml samconfig.tomlYou should see both filenames listed in the output.
File not found?
- Make sure your terminal is in the sec-analysis-pipeline folder, not inside api/ or worker/.
- Check that you saved the files with the exact names template.yaml and samconfig.toml (no extra extensions like .txt).
help me find my missing template or config file
✔️ Awesome, I've got everything!
Great. Make sure you have saved samconfig.toml.
ⓧ I'd like to double check the full code
version = 0.1
[default.deploy.parameters]
stack_name = "sec-analysis-pipeline"
resolve_s3 = true
s3_prefix = "sec-analysis-pipeline"
region = "us-east-1"
confirm_changeset = true
capabilities = "CAPABILITY_IAM"
parameter_overrides = "OpenAIApiKey=your-api-key-here"Your infrastructure is fully defined. Next up, you will build and deploy the entire stack to AWS with a single command.
Deploy and Test Your Live Pipeline
Your code is written, your infrastructure is defined in a SAM template, and your configuration file is ready. The only thing left is pushing everything to AWS so you have a live public URL.
In this step, you will build your application with the AWS SAM CLI, deploy it to AWS CloudFormation, and test your full distributed pipeline end-to-end with a real company ticker symbol.
In this step, get ready to:
- Build and deploy the full stack to AWS with SAM.
- Test the end-to-end pipeline by submitting a ticker and retrieving analysis results.
- Verify the distributed flow across all five services.
Configure your API key and deploy to AWS
Before deploying, you need to put your real OpenAI API key into the SAM configuration file. The samconfig.toml file currently has a placeholder value that SAM will pass to your worker Lambda as an environment variable.
- In VS Code, open samconfig.toml.
- Find this line:
parameter_overrides = "OpenAIApiKey=your-api-key-here"- Replace your-api-key-here with your actual OpenAI API key, so the line looks like this:
parameter_overrides = "OpenAIApiKey=sk-your-actual-openai-key"- Save samconfig.toml.
Keep your API key safe
Never commit samconfig.toml to a public Git repository after adding your key. Add it to your .gitignore if you plan to version-control this project.
Now you can build the application. SAM packages your Python code and installs the dependencies from each requirements.txt into deployment-ready artifacts.
- Build the application by running this command:
sam buildWhat does this command do?
The sam build command reads your template.yaml, finds each Lambda function's CodeUri, installs the Python dependencies from requirements.txt, and packages everything into a .aws-sam directory.
- You should see output ending with Build Succeeded.
Build failed?
Make sure you are running the command from inside the sec-analysis-pipeline folder where your template.yaml file lives.
If you see a Python-related error, confirm you have Python 3.12 installed and accessible. SAM needs a matching local runtime to install dependencies.
Help me debug my sam build error
With the build complete, deploy the stack to AWS.
- Deploy the application by running this command:
sam deployWhat happens during deploy?
SAM reads your samconfig.toml for all deployment settings (stack name, region, capabilities). It uploads your packaged code to an S3 bucket, creates a CloudFormation changeset, and provisions all the resources defined in your template.
Because confirm_changeset = true is set in your config, SAM will display a table of resources to be created and then ask you to confirm before proceeding.
- SAM will display a changeset table showing all resources to be created (the SQS queue, DynamoDB table, both Lambda functions, and the API Gateway).
- When prompted with Deploy this changeset? [y/N]: type y and press Enter.
- Wait for the deployment to finish. This takes 2-3 minutes.
- When deployment completes, SAM prints an Outputs section. Look for the ApiUrl key and copy the URL value. This is your live public API endpoint.
Save your API URL
Paste your API URL somewhere handy (a notes app or a terminal variable). You will use it for every test command in this step.
Deployment failed?
If you see an IAM-related error, make sure your AWS CLI credentials have permissions to create Lambda functions, SQS queues, DynamoDB tables, and API Gateway resources.
If the stack rolls back, check the error in the CloudFormation console. Navigate to CloudFormation in the AWS console, select the sec-analysis-pipeline stack, and check the Events tab for the failure reason.
Help me debug my SAM deploy failure
Test the pipeline end-to-end
Your API is live. Time to trigger the full distributed flow by submitting a real company ticker and watching the pipeline process it through SQS and the LLM worker.
- First, verify the API is responding by hitting the health endpoint:
curl [[API_URL="your API URL from the deploy output"]]/health- You should see a JSON response: {"status":"healthy","service":"sec-analysis-api"}.
Seeing a connection error?
Double-check that you copied the full URL from the SAM deploy output, including the https:// prefix. Do not add a trailing slash after the URL.
If you see a 500 error, your Lambda might be missing environment variables. Check that the deploy completed successfully and that your template.yaml has the correct environment variable references.
Help me debug my health endpoint error
- Submit an analysis request for Apple by running this command:
curl -X POST [[API_URL="your API URL from the deploy output"]]/analyze/AAPLWhat happens when you hit this endpoint?
Your API Lambda receives the request, looks up Apple's CIK number from the SEC database, fetches the latest 10-K filing text, stores a "processing" record in DynamoDB, and sends the filing text to your SQS queue.
The worker Lambda picks up the SQS message and sends the text to GPT-4o-mini for analysis. This whole flow takes 10-30 seconds depending on the SEC API response time and LLM processing time.
- You should see a JSON response with an analysis_id, ticker set to AAPL, and status of processing.
- Copy the analysis_id value from the response. You need it for the next command.
- Wait about 15-30 seconds for the worker to finish processing.
- Retrieve the completed analysis by running this command:
curl [[API_URL="your API URL from the deploy output"]]/results/[[ANALYSIS_ID="your analysis_id from the POST response"]]- You should see a JSON response with status set to completed and an analysis object containing fields like company_name, revenue_summary, risk_factors, and key_metrics.
Status still shows processing?
The worker Lambda needs time to call the SEC EDGAR API and then GPT-4o-mini. Wait another 15 seconds and try the GET request again.
If the status shows failed, check the error_message field in the response. Common causes are an invalid OpenAI API key or insufficient credits on your OpenAI account.
Help me debug my worker Lambda
Verify the distributed flow
You just triggered a request that flowed through five AWS services. Seeing the completed analysis proves the entire distributed pipeline is working.
- Open your browser and navigate to your API URL with /docs appended to the end.
- You should see the interactive Swagger UI showing all three endpoints: POST /analyze/{ticker}, GET /results/{analysis_id}, and GET /health.
How did the request flow through your system?
Here is the full distributed path your request traveled:
- API Gateway received your HTTP request and forwarded it to the sec-analysis-api Lambda.
- The API Lambda fetched SEC filing data, stored a processing record in DynamoDB, and sent a message to the SQS queue.
- SQS triggered the sec-analysis-worker Lambda with the message.
- The worker Lambda sent the filing text to GPT-4o-mini and wrote the structured analysis back to DynamoDB.
- Your GET request read the completed result directly from DynamoDB.
Your distributed LLM-powered pipeline is live and publicly accessible. Anyone with your URL can submit a company ticker and get AI-generated financial analysis back. Next up, the Secret Mission challenges you to extend this into a comparison engine.
Secret mission
Add a Company Comparison Endpoint
Your pipeline analyzes one company at a time. What if a recruiter wants to compare two companies side-by-side? Build a fan-out comparison endpoint that triggers parallel analysis for two tickers and generates a comparative summary.
Clean Up Your Resources
Clean Up Your Resources
Your pipeline is fully serverless, which means nothing runs (or costs money) when idle. At portfolio-demo traffic levels, your monthly AWS cost stays at $0.
Decide whether to keep your resources deployed, or delete them entirely.
Resources you used:
- API Gateway HTTP API (serves your public URL).
- Lambda function: sec-analysis-api (handles incoming requests).
- Lambda function: sec-analysis-worker (processes filings with OpenAI).
- SQS queue: sec-analysis-queue (async messaging between services).
- DynamoDB table: sec-analysis-results (stores analysis output).
- Local project files and Python virtual environment.
Why is there no Pause option?
Serverless resources like Lambda, SQS, and DynamoDB only consume resources when actively processing requests. When idle, they cost nothing and use no compute. There is nothing to pause.
Keep everything running
No action needed. Choose this if you want to keep your live API URL ready for applications or demos.
- All AWS resources remain deployed. At demo traffic (a few requests per day), your monthly cost is $0.
- Lambda, SQS, and DynamoDB are within the AWS Always Free tier (1M Lambda requests/month, 1M SQS requests/month, 25 GB DynamoDB storage).
- OpenAI API calls only cost money when you trigger a new analysis. Each analysis costs fractions of a cent.
- Keep your API URL handy for your portfolio or job applications.
Pause - I'll come back to this later
Serverless architectures have no idle state to pause. Your resources already cost $0 when not in use.
- There are no running processes to stop. Lambda functions only execute when triggered by a request.
- If you want to prevent any accidental API calls, you can delete the stack (see the Delete tab) and redeploy later with sam build && sam deploy.
Delete - I don't want to use this again
Remove all AWS resources and local project files. This is irreversible.
- Delete all AWS resources (API Gateway, both Lambda functions, SQS queue, and DynamoDB table) by running this command from your project directory:
sam delete- SAM will ask you to confirm the stack name and deletion. Enter y for each prompt.
- Remove your local project files on macOS/Linux:
cd ~/Desktop
rm -rf sec-filing-pipeline- On Windows, remove the project folder:
cd ~\Desktop
rmdir /s /q sec-filing-pipeline- Your AWS CLI and SAM CLI installations remain on your machine. These are general-purpose development tools you can reuse for other projects.
Uninstalling the CLI tools
If you also want to remove the AWS CLI and SAM CLI, follow the official uninstall guides:
That's a Wrap!
That's a Wrap!
Nice work. You just deployed a production-grade distributed system to AWS that a recruiter can interact with live.
Here's what you learned:
- Built a FastAPI REST API with interactive Swagger documentation and deployed it to AWS Lambda using SAM infrastructure-as-code.
- Implemented a distributed event-driven pipeline using Amazon SQS for async messaging between microservices.
- Integrated OpenAI GPT-4o-mini for financial document analysis, storing structured insights in DynamoDB.
- Secret Mission: Added a fan-out comparison endpoint demonstrating advanced messaging patterns.
Ready to quiz yourself?