CultivaCore API — FastAPI Template

📁 Estructura del proyecto

cultivacore-api/

app/

api/

v1/

endpoints/

├── agents.py # CRUD agentes

├── tasks.py # Tareas IA

├── clients.py # Gestión clientes

└── auth.py # OAuth2 + JWT

router.py

dependencies.py # get_current_user

core/

├── config.py # Settings + .env

├── database.py # async engine

└── security.py # JWT + bcrypt

models/ # SQLAlchemy ORM

├── agent.py · task.py

└── client.py · user.py

schemas/ # Pydantic v2

├── agent.py · task.py

└── client.py · user.py

services/ # Business logic

└── agent_service · task_service

repositories/ # Data access

└── base_repository.py + módulos

main.py # Lifespan + CORS

tests/

├── conftest.py # AsyncClient fixture

├── test_agents.py

└── test_tasks.py

Dockerfile

docker-compose.yml

pyproject.toml

.env.example

🔗 Flujo de capas

Router /api/v1 Versionado, prefijos, tags

↓

Endpoint (async def) Validates input, calls service

↓

Service (business logic) Reglas de negocio, errores

↓

Repository (data access) CRUD genérico + custom queries

↓

Model (SQLAlchemy) ORM async, mapped_column

↓

PostgreSQL (asyncpg) async engine, connection pool

📦 Stack tecnológico

FastAPI 0.111 Python 3.12 SQLAlchemy 2.x Pydantic v2 asyncpg python-jose passlib·bcrypt httpx pytest-asyncio alembic Docker uvicorn

⚙️ main.py — App entry + lifespan

app/main.py Python

from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware
from contextlib import asynccontextmanager
from app.api.v1.router import api_router
from app.core.database import engine, Base

@asynccontextmanager
async def lifespan(app: FastAPI):
    # Startup: crear tablas
    async with engine.begin() as conn:
        await conn.run_sync(Base.metadata.create_all)
    yield
    # Shutdown: dispose engine
    await engine.dispose()

app = FastAPI(
    title="CultivaCore API",
    description="Plataforma de agentes IA para CULTIVA IA",
    version="1.0.0",
    lifespan=lifespan,
    docs_url="/docs",
    redoc_url="/redoc",
)

app.add_middleware(
    CORSMiddleware,
    allow_origins=["https://cultivaia.com"],
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)

app.include_router(
    api_router,
    prefix="/api/v1"
)

🤖 models/agent.py — SQLAlchemy async

app/models/agent.py Python

from sqlalchemy import String, Boolean, Enum
from sqlalchemy.orm import Mapped, mapped_column, relationship
from app.core.database import Base
import enum

class AgentType(enum.Enum):
    MARKETING  = "marketing"
    CONTENIDO  = "contenido"
    AUTOMATIZACION = "automatizacion"
    DATOS      = "datos"
    AGENTE_IA  = "agente_ia"

class Agent(Base):
    __tablename__ = "agents"

    id: Mapped[int] = mapped_column(primary_key=True)
    name: Mapped[str] = mapped_column(String(120), nullable=False)
    slug: Mapped[str] = mapped_column(String(80), unique=True)
    model_id: Mapped[str] = mapped_column(
        String(100),
        default="claude-sonnet-4-5"
    )
    agent_type: Mapped[AgentType] = mapped_column(
        Enum(AgentType)
    )
    is_active: Mapped[bool] = mapped_column(
        Boolean, default=True
    )
    tasks: Mapped[list["Task"]] = relationship(
        back_populates="agent",
        cascade="all, delete-orphan"
    )

🛠️ services/agent_service.py

app/services/agent_service.py Python

from sqlalchemy.ext.asyncio import AsyncSession
from app.repositories.agent_repository import agent_repo
from app.schemas.agent import AgentCreate, AgentUpdate
from fastapi import HTTPException

class AgentService:
    async def register_agent(
        self,
        db: AsyncSession,
        data: AgentCreate
    ):
        # Slug must be unique
        existing = await agent_repo.get_by_slug(
            db, data.slug
        )
        if existing:
            raise HTTPException(
                status_code=409,
                detail=f"Agent '{data.slug}' ya existe"
            )
        return await agent_repo.create(db, data)

    async def deactivate_agent(
        self,
        db: AsyncSession,
        agent_id: int
    ):
        agent = await agent_repo.get(db, agent_id)
        if not agent:
            raise HTTPException(404, "Agente no encontrado")
        return await agent_repo.update(
            db, agent,
            AgentUpdate(is_active=False)
        )

agent_service = AgentService()

📡 api/v1/endpoints/agents.py

app/api/v1/endpoints/agents.py Python

from fastapi import APIRouter, Depends, status
from sqlalchemy.ext.asyncio import AsyncSession
from app.core.database import get_db
from app.schemas.agent import Agent, AgentCreate
from app.services.agent_service import agent_service
from app.api.dependencies import get_current_user

router = APIRouter(prefix="/agents", tags=["agents"])

@router.get("/", response_model=list[Agent])
async def list_agents(
    skip: int = 0, limit: int = 50,
    db: AsyncSession = Depends(get_db),
    _=Depends(get_current_user),
):
    return await agent_service.repository.get_multi(
        db, skip=skip, limit=limit
    )

@router.post(
    "/",
    response_model=Agent,
    status_code=status.HTTP_201_CREATED
)
async def create_agent(
    data: AgentCreate,
    db: AsyncSession = Depends(get_db),
    _=Depends(get_current_user),
):
    return await agent_service.register_agent(db, data)

@router.delete("/{agent_id}", status_code=204)
async def deactivate(
    agent_id: int,
    db: AsyncSession = Depends(get_db),
    _=Depends(get_current_user),
):
    await agent_service.deactivate_agent(db, agent_id)

📋 Endpoints completos — /api/v1

Método	Ruta	Descripción	Auth	Response
POST	/auth/login	OAuth2 password flow → JWT token	—	Token
POST	/auth/refresh	Renovar access token	—	Token
GET	/agents/	Listar agentes IA registrados	JWT	Agent[]
POST	/agents/	Registrar nuevo agente	JWT	Agent 201
GET	/agents/{id}	Detalle de agente por ID	JWT	Agent
PATCH	/agents/{id}	Actualizar configuración de agente	JWT	Agent
DELETE	/agents/{id}	Desactivar agente (soft delete)	JWT	204
GET	/tasks/	Listar tareas con filtro por estado	JWT	Task[]
POST	/tasks/	Crear y encolar nueva tarea IA	JWT	Task 201
GET	/tasks/{id}	Estado y output de tarea	JWT	Task
PATCH	/tasks/{id}/complete	Marcar tarea completada + guardar output	JWT	Task
GET	/clients/	Clientes de la agencia	JWT	Client[]
POST	/clients/	Alta nuevo cliente	JWT	Client 201
GET	/clients/{id}/tasks	Historial de tareas por cliente	JWT	Task[]
PATCH	/clients/{id}	Actualizar plan contratado	JWT	Client
DELETE	/clients/{id}	Dar de baja cliente	JWT	204

🧪 Test suite — pytest-asyncio

✓test_create_agent_success0.12s

✓test_create_agent_duplicate_slug0.08s

✓test_list_agents_empty0.05s

✓test_list_agents_pagination0.09s

✓test_create_task_for_agent0.14s

✓test_task_complete_saves_output0.11s

✓test_auth_login_valid0.23s

✓test_auth_login_invalid0.10s

✓test_protected_route_no_token0.04s

✓test_protected_route_expired_token0.07s

✓test_create_client0.09s

✓test_client_tasks_history0.13s

      12 passed in 1.25s ✓
    

🐳 Docker — listo para Railway/Render

Dockerfile Docker

FROM python:3.12-slim

WORKDIR /app

RUN pip install --no-cache-dir uv

COPY pyproject.toml .
RUN uv pip install --system -e .

COPY . .

CMD ["uvicorn", "app.main:app",
     "--host", "0.0.0.0",
     "--port", "8000",
     "--workers", "4"]

docker-compose.yml YAML

services:
  api:
    build: .
    ports: ["8000:8000"]
    env_file: .env
    depends_on: [db]
  db:
    image: postgres:16-alpine
    environment:
      POSTGRES_DB: cultivacore
      POSTGRES_PASSWORD: ${DB_PASSWORD}
    volumes: [pgdata:/var/lib/postgresql]
volumes:
  pgdata: