How the Securo Backend API Is Structured: FastAPI Architecture Deep Dive
Securo's backend is built on FastAPI and follows a clean modular layout that separates routing, business logic, data models, and external integrations into distinct Python packages.
The securo-finance/securo repository implements a production-grade financial API using FastAPI's dependency injection system, SQLAlchemy ORM, and Redis-backed middleware. This guide breaks down the exact file structure, request flow, and key implementation patterns found in the source code.
Core Package Structure
The backend code lives in backend/ and organizes functionality into logical layers:
| Package | Purpose | Key Files |
|---|---|---|
backend/app/main.py |
FastAPI application factory | Registers all routers, middleware, database connections |
backend/app/api/* |
HTTP endpoint routers | workspaces.py, transactions.py, users.py |
backend/app/core/* |
Cross-cutting utilities | auth.py, rate_limit.py, feature_flags.py, database.py |
backend/app/models/* |
SQLAlchemy ORM definitions | Domain entities: workspaces, assets, transactions, users |
backend/app/services/* |
Business logic layer | transaction_service.py, workspace_service.py |
backend/app/providers/* |
Third-party connectors | SimpleFin, Pluggy, OpenExchangeRates, Tesouro Direto |
backend/app/schemas/* |
Pydantic request/response models | Input validation and API documentation |
backend/mcp_server/* |
JSON-RPC server for agents | main.py, auth.py |
Each router in backend/app/api/ defines a URL prefix (e.g., /api/workspaces) and tags for OpenAPI documentation grouping.
Request Lifecycle: From HTTP to Database
Understanding how requests flow through the Securo backend API reveals why the modular structure matters.
Step 1: FastAPI Application Entry Point
In backend/app/main.py, the application factory creates the main FastAPI instance and attaches global middleware:
# Simplified representation based on source structure
from fastapi import FastAPI
from app.api import workspaces, transactions, users
from app.core import auth, rate_limit
app = FastAPI(title="Securo API")
# Global middleware: CORS, rate limiting, authentication
app.add_middleware(rate_limit.RateLimitMiddleware)
app.add_middleware(auth.AuthMiddleware)
# Register all API routers with prefixes
app.include_router(workspaces.router, prefix="/api/workspaces", tags=["workspaces"])
app.include_router(transactions.router, prefix="/api/transactions", tags=["transactions"])
Step 2: Router Dispatch and Dependency Injection
When a request hits GET /api/workspaces, FastAPI matches it to the workspace router. Handlers declare dependencies that resolve automatically:
# backend/app/api/workspaces.py (conceptual structure)
from fastapi import APIRouter, Depends
from app.core.auth import get_current_user
from app.core.workspace_context import workspace_context
from app.services.workspace_service import WorkspaceService
@router.get("/")
async def list_workspaces(
user: User = Depends(get_current_user), # JWT validation
ctx: WorkspaceContext = Depends(workspace_context) # DB session + workspace scoping
):
service = WorkspaceService(ctx.db)
return await service.get_user_workspaces(user.id)
The workspace_context dependency (from backend/app/core/workspace_context.py) is critical: it injects the current workspace scope and database session, ensuring multi-tenant data isolation.
Step 3: Service Layer Execution
Business logic lives in backend/app/services/* classes. Services encapsulate complex operations and remain agnostic to HTTP concerns:
# Calling the service layer directly (internal Python usage)
from backend.app.services.transaction_service import TransactionService
from backend.app.schemas.transaction import TransactionCreate
from backend.app.core.database import get_db
async def create_example_txn():
async with get_db() as db: # scoped DB session from database.py
service = TransactionService(db)
txn = await service.create(
TransactionCreate(
account_id="acc_123",
amount=123.45,
date="2024-09-01",
description="Coffee",
currency="USD",
)
)
return txn
Step 4: Provider Abstraction for External Data
When services need external data, they call provider classes in backend/app/providers/*. Each provider implements a uniform async interface:
SimpleFinProvider– Bank account aggregationPluggyProvider– Latin American financial dataOpenExchangeRatesProvider– Currency conversionTesouroDiretoProvider– Brazilian treasury bonds
This abstraction lets services work with financial data without hardcoding third-party API details.
Step 5: ORM Persistence
Services use SQLAlchemy models from backend/app/models/* within scoped sessions managed by backend/app/core/database.py. The session context ensures proper transaction boundaries and connection pooling.
Authentication and Security
Securo implements JWT/OIDC authentication through FastAPI-Users integrated in backend/app/core/auth.py.
Protected routes depend on fastapi_users.current_user():
from fastapi_users import fastapi_users
@app.get("/api/protected")
async def protected_route(user: User = Depends(fastapi_users.current_user())):
return {"user_id": user.id}
The backend/app/api/oidc_auth.py router handles OIDC flow initiation and callback, enabling enterprise SSO integrations.
Rate Limiting and Feature Flags
Two core utilities enforce operational policies:
| Utility | File | Function |
|---|---|---|
| Rate limiting | backend/app/core/rate_limit.py |
Redis-backed token bucket; applied as middleware or per-route dependency |
| Feature flags | backend/app/core/feature_flags.py |
Gradual rollout of new API capabilities |
Example rate-limited endpoint:
from app.core.rate_limit import rate_limit
@app.post("/api/expensive-operation")
async def expensive_op(
_rate_limit: None = Depends(rate_limit(requests=10, window=60))
):
pass
MCP Server: JSON-RPC for Agent Communication
The backend/mcp_server/ package runs a separate FastAPI application for JSON-RPC 2.0 communication with Securo agents.
In backend/mcp_server/main.py:
app = FastAPI(title="Securo MCP Server", openapi_url=None, docs_url=None)
# Only registers the JSON-RPC route
app.post("/mcp")(mcp_endpoint)
Key characteristics:
- Low latency – Bypasses OpenAPI validation overhead
- Agent-native – Agents call methods like
agents.knowledge.searchdirectly - Same process – Shares database connections with the main API
Example client call to the MCP endpoint:
import httpx, json, asyncio
async def mcp_call(method: str, params: dict):
payload = {"jsonrpc": "2.0", "id": 1, "method": method, "params": params}
async with httpx.AsyncClient(base_url="https://api.securo.dev") as client:
resp = await client.post("/mcp", json=payload)
resp.raise_for_status()
return resp.json()["result"]
# Trigger agent knowledge search
# result = asyncio.run(mcp_call("agents.knowledge.search", {"query": "tax deductible"}))
Authentication for MCP endpoints is handled separately in backend/mcp_server/auth.py with JWT validation.
Complete API Client Example
Here's a runnable pattern for calling Securo's REST API from Python:
import httpx
import asyncio
async def list_workspaces(token: str):
async with httpx.AsyncClient(base_url="https://api.securo.dev") as client:
resp = await client.get(
"/api/workspaces",
headers={"Authorization": f"Bearer {token}"}
)
resp.raise_for_status()
return resp.json() # → list of workspace dicts
# Example usage
# workspaces = asyncio.run(list_workspaces("<jwt-token>"))
Key Source Files Reference
| File | Role |
|---|---|
backend/app/main.py |
Application factory; middleware and router registration |
backend/app/api/workspaces.py |
Workspace CRUD operations |
backend/app/api/transactions.py |
Transaction management, bulk import, split detection |
backend/app/core/auth.py |
FastAPI-Users JWT/OIDC integration |
backend/app/core/workspace_context.py |
Multi-tenant dependency injection |
backend/app/core/rate_limit.py |
Redis token-bucket implementation |
backend/app/services/transaction_service.py |
Core transaction business logic |
backend/mcp_server/main.py |
JSON-RPC server for agent communication |
Summary
- FastAPI modularity – Securo separates routing (
api/), utilities (core/), business logic (services/), and data access (models/,providers/) into distinct packages - Dependency injection –
workspace_context,get_current_user, andrate_limitdependencies enforce cross-cutting concerns without cluttering handler code - Dual protocol support – Standard REST OpenAPI at
/api/*plus JSON-RPC at/mcpfor specialized agent communication - Production middleware – Redis-backed rate limiting, JWT authentication, and feature flags are core utilities reusable across all routers
- Provider pattern – External financial data sources are abstracted behind uniform async interfaces in
backend/app/providers/
Frequently Asked Questions
What framework does Securo use for its backend API?
Securo uses FastAPI as its primary web framework, as implemented in backend/app/main.py. The architecture leverages FastAPI's native dependency injection system, automatic OpenAPI documentation generation, and async/await support for high-performance I/O operations with databases and external providers.
How does Securo handle multi-tenant workspace isolation?
Workspace isolation is enforced through the workspace_context dependency in backend/app/core/workspace_context.py. This dependency injects both a database session and the current workspace scope into route handlers, ensuring all queries are automatically filtered to the authenticated user's workspace without repetitive code in every endpoint.
What is the MCP server and why does Securo need it?
The MCP server is a JSON-RPC 2.0 endpoint running at /mcp from backend/mcp_server/main.py. It provides a low-overhead protocol specifically for Securo agents to execute remote procedure calls like agents.knowledge.search. Unlike the REST API, it disables OpenAPI validation (openapi_url=None) to minimize latency for high-frequency agent interactions while still sharing database connections and authentication infrastructure with the main application.
Where does Securo implement rate limiting?
Rate limiting is implemented in backend/app/core/rate_limit.py using a Redis-backed token bucket algorithm. It can be applied globally through middleware or selectively per route via Depends(rate_limit(...)), allowing different endpoints to have customized request quotas based on their computational cost and business criticality.
Have a question about this repo?
These articles cover the highlights, but your codebase questions are specific. Give your agent direct access to the source. Share this with your agent to get started:
curl -s "https://instagit.com/install.md" Maintain an open-source project? Get it listed too →