FastAPI Agent Blueprint

AI Agent Backend Platform on FastAPI.
MCP server + AI orchestration + async DDD — from CRUD to agent tools in one codebase.
Build AI-powered backends that serve both users and agents.

Quick Start · Architecture · AI-Native Development · Comparison · 한국어

Key Features

AI Agent Platform

• MCP Server interface planned — Expose domain services as AI agent tools via FastMCP
• AI agent orchestration planned — PydanticAI integration for structured LLM workflows
• Vector search planned — pgvector for semantic search, RAG, and similarity matching

Production-Ready Architecture

• 4 interface types — HTTP API (FastAPI) + Async Worker (Taskiq) + Admin UI (SQLAdmin) + MCP Server (planned)
• Zero-boilerplate CRUD — Inherit BaseRepository[DTO] + BaseService[DTO], get 7 async CRUD methods instantly
• Auto domain discovery — Add a domain folder, it auto-registers. No container or bootstrap changes needed
• Async-first — Genuine async from DB (asyncpg) to HTTP (aiohttp) to task queue (Taskiq)

Developer Experience

• 14 AI development skills — Claude Code slash commands for scaffolding, testing, architecture review, and more
• Architecture enforcement — Pre-commit hooks block Domain -> Infrastructure imports at commit time
• Type-safe generics — BaseRepository[ProductDTO], BaseService[ProductDTO], SuccessResponse[ProductResponse]
• DDD layered structure — Each domain is fully independent with its own layers (Domain / Infrastructure / Interface / Application)
• 14 Architecture Decision Records — Every major design choice documented with rationale

Why?

Your domain logic — accessible everywhere

Write business logic once. Expose it as a REST API, background job, admin view, or AI agent tool.

# 1. Define your service
class DocumentService(BaseService[DocumentDTO]):
    async def analyze(self, document_id: int) -> AnalysisDTO:
        ...  # your business logic

# 2. REST API — for your frontend
@router.post("/documents/{document_id}/analyze")
async def analyze_document(document_id: int, service=Depends(...)):
    return await service.analyze(document_id)

# 3. MCP Tool — for AI agents (planned)
@mcp.tool()
async def analyze_document(document_id: int) -> AnalysisResult:
    return await document_service.analyze(document_id)

# 4. Background job — for batch processing
@broker.task()
async def batch_analyze(project_id: int):
    for doc in await service.get_by_project(project_id):
        await service.analyze(doc.id)

Zero-boilerplate CRUD

# Before: Repeat the same CRUD for every domain
@router.post("/user")
async def create_user(user: UserCreate):
    db = get_db()
    new_user = User(**user.dict())
    db.add(new_user)
    db.commit()
    return new_user

@router.post("/product")  # Repeat again...
async def create_product(product: ProductCreate):
    db = get_db()
    new_product = Product(**product.dict())
    db.add(new_product)
    db.commit()
    return new_product

# After: One inheritance line, CRUD done
class ProductRepository(BaseRepository[ProductDTO]):
    def __init__(self, database: Database):
        super().__init__(database=database, model=ProductModel, return_entity=ProductDTO)

class ProductService(BaseService[ProductDTO]):
    def __init__(self, product_repository: ProductRepositoryProtocol):
        super().__init__(repository=product_repository)

# 7 CRUD methods provided automatically -- just add your custom logic

Architecture

Router -> Service(BaseService) -> Repository(BaseRepository) -> DB
               ^ Simple CRUD: this is all you need

Router -> UseCase -> Service -> Repository -> DB
               ^ Add only when complex business logic is required

Layer Responsibilities

Data Flow

Write: Request --> Service --> Repository --> Model -> DB
Read:  Response <-- Service <-- Repository <-- DTO <-- Model

• Request is passed directly to Service (no extra conversion needed)
• Repository handles Model -> DTO conversion (model_validate(model, from_attributes=True))
• Router handles DTO -> Response conversion (Response(**dto.model_dump(exclude={...})))

Data Objects

AI-Native Development (AIDD)

This template works great on its own. With Claude Code, it's magic.

Zero Learning Curve

Complex architecture? Type /onboard -- it explains everything at your level.

The /onboard skill adapts to your experience:

• Beginner: Explains DDD concepts with simple analogies
• Intermediate: Focuses on this project's specific patterns
• Advanced: Jumps straight to architecture rules and conventions

14 Built-in Skills

MCP Server Setup

To use AIDD features, configure these MCP servers:

Serena -- Symbolic code navigation/editing (LSP-level rename, reference analysis)

{
  "mcpServers": {
    "serena": {
      "command": "uvx",
      "args": ["--from", "serena-mcp", "serena", "--project-root", "."]
    }
  }
}

context7 -- Up-to-date library documentation

{
  "mcpServers": {
    "context7": {
      "command": "npx",
      "args": ["-y", "@upstash/context7-mcp@latest"]
    }
  }
}

The project works without MCP servers. AIDD skills require MCP server configuration.

Quick Start

# 1. Clone
git clone https://github.com/Mr-DooSun/fastapi-agent-blueprint.git
cd fastapi-agent-blueprint

# 2. Setup (requires uv)
make setup

# 3. Set up environment variables
cp _env/local.env.example _env/local.env

# 4. Start PostgreSQL + run migrations + start server
make dev

Open http://localhost:8000/docs-swagger to explore the API.

# 2. Create virtual environment + install dependencies
uv venv --python 3.12
source .venv/bin/activate
uv sync --group dev

# 3. Set up environment variables
cp _env/local.env.example _env/local.env

# 4. Start PostgreSQL (Docker)
docker compose -f docker-compose.local.yml up -d postgres

# 5. Run migrations + start server
alembic upgrade head
python run_server_local.py --env local

Adding a New Domain

Using Product domain as an example:

1. Domain Layer

# src/product/domain/dtos/product_dto.py
class ProductDTO(BaseModel):
    id: int = Field(..., description="Product ID")
    name: str = Field(..., description="Product name")
    price: int = Field(..., description="Price")
    created_at: datetime
    updated_at: datetime

# src/product/domain/protocols/product_repository_protocol.py
class ProductRepositoryProtocol(BaseRepositoryProtocol[ProductDTO]):
    pass

# src/product/domain/services/product_service.py
class ProductService(BaseService[ProductDTO]):
    def __init__(self, product_repository: ProductRepositoryProtocol):
        super().__init__(repository=product_repository)
    # CRUD provided automatically. Just add custom logic.

2. Infrastructure Layer

# src/product/infrastructure/database/models/product_model.py
class ProductModel(Base):
    __tablename__ = "product"
    id: Mapped[int] = mapped_column(Integer, primary_key=True)
    name: Mapped[str] = mapped_column(String(255), nullable=False)
    price: Mapped[int] = mapped_column(Integer, nullable=False)
    created_at: Mapped[datetime] = mapped_column(DateTime, server_default=func.now())
    updated_at: Mapped[datetime] = mapped_column(DateTime, onupdate=func.now())

# src/product/infrastructure/repositories/product_repository.py
class ProductRepository(BaseRepository[ProductDTO]):
    def __init__(self, database: Database):
        super().__init__(database=database, model=ProductModel, return_entity=ProductDTO)

# src/product/infrastructure/di/product_container.py
class ProductContainer(containers.DeclarativeContainer):
    core_container = providers.DependenciesContainer()
    product_repository = providers.Singleton(ProductRepository, database=core_container.database)
    product_service = providers.Factory(ProductService, product_repository=product_repository)

3. Interface Layer

# src/product/interface/server/routers/product_router.py
@router.post("/product", response_model=SuccessResponse[ProductResponse])
@inject
async def create_product(
    item: CreateProductRequest,
    product_service: ProductService = Depends(Provide[ProductContainer.product_service]),
) -> SuccessResponse[ProductResponse]:
    data = await product_service.create_data(entity=item)
    return SuccessResponse(data=ProductResponse(**data.model_dump()))

Auto Registration

discover_domains() automatically detects new domains.
No changes needed in _apps/ containers or bootstrap files.

Discovery conditions:

• src/{name}/__init__.py exists
• src/{name}/infrastructure/di/{name}_container.py exists

With Claude Code, just run /new-domain product to scaffold all of this automatically.

4 Interface Types

Each domain can expose functionality through multiple interfaces:

All interfaces share the same Domain and Infrastructure layers -- write your business logic once, expose it everywhere.

Tech Stack

AI & Agent planned

Core

Infrastructure

DevOps

Project Structure

src/
├── _apps/                        # App entry points
│   ├── server/                  # FastAPI HTTP server
│   ├── worker/                  # Taskiq async worker
│   └── admin/                   # SQLAdmin dashboard
│
├── _core/                        # Shared infrastructure
│   ├── domain/
│   │   ├── protocols/           # BaseRepositoryProtocol[ReturnDTO]
│   │   └── services/            # BaseService[ReturnDTO]
│   ├── infrastructure/
│   │   ├── database/            # Database, BaseRepository[ReturnDTO]
│   │   ├── http/                # HttpClient, BaseHttpGateway
│   │   ├── taskiq/              # SQS Broker, TaskiqManager
│   │   ├── storage/             # S3/MinIO
│   │   ├── di/                  # CoreContainer
│   │   └── discovery.py         # Auto domain discovery
│   ├── application/dtos/        # BaseRequest, BaseResponse, SuccessResponse
│   ├── exceptions/              # Exception handlers, BaseCustomException
│   └── config.py                # Settings (pydantic-settings)
│
├── user/                         # Example domain
│   ├── domain/
│   │   ├── dtos/                # UserDTO
│   │   ├── protocols/           # UserRepositoryProtocol
│   │   ├── services/            # UserService(BaseService[UserDTO])
│   │   ├── exceptions/          # UserNotFoundException
│   │   └── events/              # UserCreated, UserUpdated
│   ├── infrastructure/
│   │   ├── database/models/     # UserModel
│   │   ├── repositories/        # UserRepository(BaseRepository[UserDTO])
│   │   └── di/                  # UserContainer
│   └── interface/
│       ├── server/              # routers/, schemas/, bootstrap/
│       ├── worker/              # tasks/, bootstrap/
│       └── admin/               # SQLAdmin views
│
├── migrations/                   # Alembic
├── _env/                         # Environment variables
└── docs/history/                 # Architecture Decision Records

Comparison

Architecture Decisions

Every technical choice in this project is documented as an ADR (Architecture Decision Record).

View all 14 ADRs

Roadmap

Phase 1: AI Agent Foundation

• FastMCP interface (#18)
• PydanticAI integration (#15)
• pgvector support (#11)
• JWT authentication (#4)

Phase 2: Production Readiness

• Structured logging — structlog (#9)
• Per-environment config (#7, #8, #16)
• Error notifications (#17)
• Worker payload schemas (#37)
• CRUD data validation (#10)

Phase 3: Ecosystem

• Test coverage expansion (#2)
• Performance testing — Locust (#3)
• Serverless deployment (#6)
• WebSocket documentation (#1)
• CHANGELOG (#41)

Completed

• Health check endpoint
• Auto domain discovery
• Architecture enforcement (pre-commit)
• 14 AI development skills

Star this repo to follow our progress!

Contributing

See CONTRIBUTING.md for development setup, coding guidelines, and PR process.

License

MIT License -- Free for commercial use, modification, and distribution.

Star History