This article is published in English.
Production AI Agents with FastAPI, LangGraph, and Clean Architecture
Layer boundaries, typed graph state, testable services, and deployment shape that keeps agents maintainable.
This walkthrough rebuilds the path from raw materials to a working system for: How I Structure Production AI Agents with FastAPI, LangGraph and Clean Architecture. The focus is operable steps, explicit checks, and code that you can drop into a repo without guessing intent. For Overview, define the inputs, the owner of the step, and the exit criteria before changing code. Operators should be able to re-run the step from a known checkpoint without guessing hidden state. Keep configuration outside application code. Environment files, secret stores, and feature flags belong in one place operators can audit without reading the whole graph.
@router.post("/suppliers")
async def create_supplier(request: SupplierRequest):
policies = opensearch.search(
index="supplier-policies",
query=request.description,
)
response = bedrock.converse(
modelId=MODEL_ID,
messages=build_messages(request, policies),
)
supplier = Supplier(
name=request.name,
tax_id=request.tax_id,
)
db.add(supplier)
db.commit()
sqs.send_message(
QueueUrl=SUPPLIER_QUEUE,
MessageBody=serialize(supplier),
)
return {"status": "created"}
FastAPI Is an Interface — Not the Application
For FastAPI Is an Interface — Not the Application, define the inputs, the owner of the step, and the exit criteria before changing code. Operators should be able to re-run the step from a known checkpoint without guessing hidden state. Document the happy path and the recovery path together. Retries, human gates, and dead-letter handling are part of the product, not later polish. Use constraints and indexes before bulk ingest. Uniqueness on business keys turns later merges into predictable upserts instead of duplicate storms.
@router.post("/suppliers")
async def create_supplier(
request: CreateSupplierRequest,
use_case: CreateSupplierUseCase = Depends(
get_create_supplier_use_case
),
):
command = CreateSupplierCommand(
name=request.name,
tax_id=request.tax_id,
country=request.country,
)
result = await use_case.execute(command)
return CreateSupplierResponse.from_result(result)
HTTP Request
↓
FastAPI
↓
Application
↓
Domain
↓
Ports
↓
Adapters
The Application Talks to Capabilities, Not Technologies
For The Application Talks to Capabilities, Not Technologies, define the inputs, the owner of the step, and the exit criteria before changing code. Operators should be able to re-run the step from a known checkpoint without guessing hidden state. Prefer small, testable units over sprawling scripts. When a step fails, the failure should point at a single responsibility rather than a tangled pipeline. Use constraints and indexes before bulk ingest. Uniqueness on business keys turns later merges into predictable upserts instead of duplicate storms.
class CreateSupplierUseCase:
def __init__(
self,
repository: SupplierRepository,
policy_service: SupplierPolicyService,
event_publisher: EventPublisher,
):
self.repository = repository
self.policy_service = policy_service
self.event_publisher = event_publisher
async def execute(
self,
command: CreateSupplierCommand,
) -> Supplier:
existing = await self.repository.find_by_tax_id(
command.tax_id
)
if existing:
raise SupplierAlreadyExists(command.tax_id)
policy = await self.policy_service.evaluate(
country=command.country
)
supplier = Supplier.create(
name=command.name,
tax_id=command.tax_id,
country=command.country,
requires_approval=policy.requires_approval,
)
await self.repository.add(supplier)
await self.event_publisher.publish(
SupplierCreated(supplier.id)
)
return supplier
Ports Create the Boundary
For Ports Create the Boundary, define the inputs, the owner of the step, and the exit criteria before changing code. Operators should be able to re-run the step from a known checkpoint without guessing hidden state. Treat this stage as a contract between inputs and validated outputs. Name the artifacts, define success checks, and refuse silent partial completion. Use constraints and indexes before bulk ingest. Uniqueness on business keys turns later merges into predictable upserts instead of duplicate storms. For Ports Create the Boundary, define the inputs, the owner of the step, and the exit criteria before changing code. Operators should be able to re-run the step from a known checkpoint without guessing hidden state. Keep configuration outside application code. Environment files, secret stores, and feature flags belong in one place operators can audit without reading the whole graph.
from typing import Protocol
class SupplierRepository(Protocol):
async def find_by_tax_id(
self,
tax_id: str,
) -> Supplier | None:
...
async def add(
self,
supplier: Supplier,
) -> None:
...
class PolicyRetriever(Protocol):
async def retrieve(
self,
query: str,
) -> list[PolicyDocument]:
...
class LLMProvider(Protocol):
async def reason(
self,
context: AgentContext,
) -> AgentDecision:
...
class EventPublisher(Protocol):
async def publish(
self,
event: DomainEvent,
) -> None:
...
SupplierRepository
PolicyRetriever
LLMProvider
EventPublisher
PostgreSQL
Amazon OpenSearch
Amazon Bedrock
AWS SQS
Where Amazon Bedrock Belongs
For Where Amazon Bedrock Belongs, define the inputs, the owner of the step, and the exit criteria before changing code. Operators should be able to re-run the step from a known checkpoint without guessing hidden state. Document the happy path and the recovery path together. Retries, human gates, and dead-letter handling are part of the product, not later polish. Inspect a sample neighborhood after the first batch. Cypher probes catch label drift and missing properties cheaper than end-to-end chat tests.
class BedrockLLMProvider(LLMProvider):
def __init__(self, client, model_id: str):
self.client = client
self.model_id = model_id
async def reason(
self,
context: AgentContext,
) -> AgentDecision:
response = self.client.converse(
modelId=self.model_id,
messages=build_messages(context),
)
return map_bedrock_response(response)
Application
↓
LLMProvider
↑
BedrockLLMProvider
Where RAG Belongs
For Where RAG Belongs, define the inputs, the owner of the step, and the exit criteria before changing code. Operators should be able to re-run the step from a known checkpoint without guessing hidden state. Prefer small, testable units over sprawling scripts. When a step fails, the failure should point at a single responsibility rather than a tangled pipeline. Inspect a sample neighborhood after the first batch. Cypher probes catch label drift and missing properties cheaper than end-to-end chat tests.
FastAPI
↓
OpenSearch
↓
LLM
class OpenSearchPolicyRetriever(PolicyRetriever):
def __init__(
self,
opensearch_client,
embedding_provider,
):
self.client = opensearch_client
self.embedding_provider = embedding_provider
async def retrieve(
self,
query: str,
) -> list[PolicyDocument]:
vector = await self.embedding_provider.embed(query)
results = self.client.search(
index="supplier-policies",
body=build_vector_query(vector),
)
return map_documents(results)
RAG Gives Context. It Doesn’t Own the Decision.
For RAG Gives Context. It Doesn’t Own the Decision., define the inputs, the owner of the step, and the exit criteria before changing code. Operators should be able to re-run the step from a known checkpoint without guessing hidden state. Treat this stage as a contract between inputs and validated outputs. Name the artifacts, define success checks, and refuse silent partial completion. Inspect a sample neighborhood after the first batch. Cypher probes catch label drift and missing properties cheaper than end-to-end chat tests. For RAG Gives Context. It Doesn’t Own the Decision., define the inputs, the owner of the step, and the exit criteria before changing code. Operators should be able to re-run the step from a known checkpoint without guessing hidden state. Keep configuration outside application code. Environment files, secret stores, and feature flags belong in one place operators can audit without reading the whole graph.
RAG
↓
LLM
↓
"Looks fine"
↓
Create Supplier
RAG
↓
Relevant Context
↓
Agent Reasoning
↓
Application
↓
Domain Rules
↓
Decision
Where LangGraph Belongs
For Where LangGraph Belongs, define the inputs, the owner of the step, and the exit criteria before changing code. Operators should be able to re-run the step from a known checkpoint without guessing hidden state. Document the happy path and the recovery path together. Retries, human gates, and dead-letter handling are part of the product, not later polish. Model nodes and relationships for the questions you will ask, not for every noun in the document. Sparse, typed edges beat dense mystery graphs.
Understand Request
↓
Retrieve Policies
↓
Evaluate Information
↓
Need More Data?
↙ ↘
Yes No
↓ ↓
Ask User Continue
↓
Approval Required?
↙ ↘
Yes No
↓ ↓
Human Approval Continue
↘ ↙
Request Action
graph = StateGraph(AgentState)
graph.add_node(
"understand_intent",
understand_intent,
)
graph.add_node(
"retrieve_policies",
retrieve_policies,
)
graph.add_node(
"evaluate",
evaluate_request,
)
graph.add_node(
"human_approval",
request_human_approval,
)
graph.add_node(
"request_action",
request_application_action,
)
graph.add_edge(
"understand_intent",
"retrieve_policies",
)
graph.add_edge(
"retrieve_policies",
"evaluate",
)
graph.add_conditional_edges(
"evaluate",
determine_next_step,
{
"approval": "human_approval",
"execute": "request_action",
},
)
LangGraph Shouldn’t Become the New Monolith
For LangGraph Shouldn’t Become the New Monolith, define the inputs, the owner of the step, and the exit criteria before changing code. Operators should be able to re-run the step from a known checkpoint without guessing hidden state. Prefer small, testable units over sprawling scripts. When a step fails, the failure should point at a single responsibility rather than a tangled pipeline. Model nodes and relationships for the questions you will ask, not for every noun in the document. Sparse, typed edges beat dense mystery graphs.
def evaluate_node(state):
if state.amount > 100_000:
state.requires_approval = True
if state.country == "BR":
...
if state.supplier_type == "CRITICAL":
...
decision = supplier_policy.evaluate(
supplier=supplier,
context=context,
)
LangGraph → orchestrates
Domain → decides
Where MCP Fits
For Where MCP Fits, define the inputs, the owner of the step, and the exit criteria before changing code. Operators should be able to re-run the step from a known checkpoint without guessing hidden state. Treat this stage as a contract between inputs and validated outputs. Name the artifacts, define success checks, and refuse silent partial completion. Model nodes and relationships for the questions you will ask, not for every noun in the document. Sparse, typed edges beat dense mystery graphs. For Where MCP Fits, define the inputs, the owner of the step, and the exit criteria before changing code. Operators should be able to re-run the step from a known checkpoint without guessing hidden state. Keep configuration outside application code. Environment files, secret stores, and feature flags belong in one place operators can audit without reading the whole graph.
LLM
↓
PostgreSQL
LLM
↓
SAP
Agent
↓
MCP Tool
↓
Application API
↓
Use Case
↓
Domain
↓
Infrastructure
create_supplier
execute_sql
PostgreSQL Is Also an Adapter
For PostgreSQL Is Also an Adapter, define the inputs, the owner of the step, and the exit criteria before changing code. Operators should be able to re-run the step from a known checkpoint without guessing hidden state. Document the happy path and the recovery path together. Retries, human gates, and dead-letter handling are part of the product, not later polish. Use constraints and indexes before bulk ingest. Uniqueness on business keys turns later merges into predictable upserts instead of duplicate storms.
class SupplierRepository(Protocol):
async def add(
self,
supplier: Supplier,
) -> None:
...
class PostgresSupplierRepository(
SupplierRepository
):
def __init__(self, session):
self.session = session
async def add(
self,
supplier: Supplier,
) -> None:
entity = SupplierModel.from_domain(
supplier
)
self.session.add(entity)
FastAPI → doesn't know PostgreSQL exists
Application → doesn't know PostgreSQL existsDomain → doesn't know PostgreSQL existsInfrastructure → does
SAP Should Be Behind Another Boundary
For SAP Should Be Behind Another Boundary, define the inputs, the owner of the step, and the exit criteria before changing code. Operators should be able to re-run the step from a known checkpoint without guessing hidden state. Prefer small, testable units over sprawling scripts. When a step fails, the failure should point at a single responsibility rather than a tangled pipeline. Use constraints and indexes before bulk ingest. Uniqueness on business keys turns later merges into predictable upserts instead of duplicate storms.
CreateSupplierUseCase
↓
Persist State
↓
SupplierCreated
↓
AWS SQS
↓
Integration Worker
↓
SAP
Putting the Architecture Together
For Putting the Architecture Together, define the inputs, the owner of the step, and the exit criteria before changing code. Operators should be able to re-run the step from a known checkpoint without guessing hidden state. Treat this stage as a contract between inputs and validated outputs. Name the artifacts, define success checks, and refuse silent partial completion. Use constraints and indexes before bulk ingest. Uniqueness on business keys turns later merges into predictable upserts instead of duplicate storms. For Putting the Architecture Together, define the inputs, the owner of the step, and the exit criteria before changing code. Operators should be able to re-run the step from a known checkpoint without guessing hidden state. Keep configuration outside application code. Environment files, secret stores, and feature flags belong in one place operators can audit without reading the whole graph.
USER
│
▼
FastAPI
│
▼
LangGraph
│
┌────────────┼────────────┐
│ │ │
▼ ▼ ▼
RAG LLM MCP
│ │ │
▼ ▼ │
OpenSearch Bedrock │
▼
Application
│
▼
Domain
│
┌─────────┴─────────┐
│ │
▼ ▼
PostgreSQL AWS SQS
│
▼
Worker
│
▼
SAP
Two Worlds
For Two Worlds, define the inputs, the owner of the step, and the exit criteria before changing code. Operators should be able to re-run the step from a known checkpoint without guessing hidden state. Document the happy path and the recovery path together. Retries, human gates, and dead-letter handling are part of the product, not later polish. Inspect a sample neighborhood after the first batch. Cypher probes catch label drift and missing properties cheaper than end-to-end chat tests.
PROBABILISTICLLM
RAG
Natural Language
Intent Understanding
Agent Reasoning
Tool Selection
DETERMINISTIC
Authorization
Business Rules
Transactions
Persistence
Idempotency
Integration
Auditability
A Practical Test for the Architecture
For A Practical Test for the Architecture, define the inputs, the owner of the step, and the exit criteria before changing code. Operators should be able to re-run the step from a known checkpoint without guessing hidden state. Prefer small, testable units over sprawling scripts. When a step fails, the failure should point at a single responsibility rather than a tangled pipeline. Inspect a sample neighborhood after the first batch. Cypher probes catch label drift and missing properties cheaper than end-to-end chat tests.
async def test_supplier_requires_approval():
repository = FakeSupplierRepository()
policies = FakePolicyService(
requires_approval=True
)
events = FakeEventPublisher()
use_case = CreateSupplierUseCase(
repository=repository,
policy_service=policies,
event_publisher=events,
)
supplier = await use_case.execute(
CreateSupplierCommand(
name="ACME",
tax_id="123",
country="BR",
)
)
assert supplier.requires_approval is True
Changing the Model Should Be Boring
For Changing the Model Should Be Boring, define the inputs, the owner of the step, and the exit criteria before changing code. Operators should be able to re-run the step from a known checkpoint without guessing hidden state. Treat this stage as a contract between inputs and validated outputs. Name the artifacts, define success checks, and refuse silent partial completion. Inspect a sample neighborhood after the first batch. Cypher probes catch label drift and missing properties cheaper than end-to-end chat tests. For Changing the Model Should Be Boring, define the inputs, the owner of the step, and the exit criteria before changing code. Operators should be able to re-run the step from a known checkpoint without guessing hidden state. Keep configuration outside application code. Environment files, secret stores, and feature flags belong in one place operators can audit without reading the whole graph.
LLMProvider
▲
│
┌─────────┴─────────┐
│ │
BedrockProvider AnotherProvider
FastAPI
Application
Domain
Business Rules
Clean Architecture Is Not a Folder Structure
For Clean Architecture Is Not a Folder Structure, define the inputs, the owner of the step, and the exit criteria before changing code. Operators should be able to re-run the step from a known checkpoint without guessing hidden state. Document the happy path and the recovery path together. Retries, human gates, and dead-letter handling are part of the product, not later polish. Model nodes and relationships for the questions you will ask, not for every noun in the document. Sparse, typed edges beat dense mystery graphs.
src/
├── api/
├── application/
├── domain/
├── agents/
├── ports/
└── infrastructure/
AI Makes These Boundaries More Important
For AI Makes These Boundaries More Important, define the inputs, the owner of the step, and the exit criteria before changing code. Operators should be able to re-run the step from a known checkpoint without guessing hidden state. Prefer small, testable units over sprawling scripts. When a step fails, the failure should point at a single responsibility rather than a tangled pipeline. Model nodes and relationships for the questions you will ask, not for every noun in the document. Sparse, typed edges beat dense mystery graphs.
The Architecture in One Sentence
For The Architecture in One Sentence, define the inputs, the owner of the step, and the exit criteria before changing code. Operators should be able to re-run the step from a known checkpoint without guessing hidden state. Treat this stage as a contract between inputs and validated outputs. Name the artifacts, define success checks, and refuse silent partial completion. Model nodes and relationships for the questions you will ask, not for every noun in the document. Sparse, typed edges beat dense mystery graphs. For The Architecture in One Sentence, define the inputs, the owner of the step, and the exit criteria before changing code. Operators should be able to re-run the step from a known checkpoint without guessing hidden state. Keep configuration outside application code. Environment files, secret stores, and feature flags belong in one place operators can audit without reading the whole graph.
Final Thought
For Final Thought, define the inputs, the owner of the step, and the exit criteria before changing code. Operators should be able to re-run the step from a known checkpoint without guessing hidden state. Document the happy path and the recovery path together. Retries, human gates, and dead-letter handling are part of the product, not later polish. Use constraints and indexes before bulk ingest. Uniqueness on business keys turns later merges into predictable upserts instead of duplicate storms.
Prompt
↓
LLM
↓
Tool
↓
Database