Backend Architecture

FastAPI application structure, API routes, middleware, and service orchestration

Application Structure

The backend is a FastAPI application (app/main.py) with async/await throughout. All I/O operations (database, Neo4j, HTTP calls) are non-blocking.

app/
├── main.py                       # FastAPI app entry, lifespan, CORS, middleware
├── core/
│   ├── config.py                 # Pydantic settings (all env vars)
│   ├── database.py               # Async SQLAlchemy engine + session
│   ├── neo4j_client.py           # Neo4j async driver singleton
│   └── neo4j_schema.py           # Constraints + indexes
├── models/
│   ├── user.py                   # GitHub users
│   ├── installation.py           # Connected repos + webhook secrets
│   ├── event.py                  # Incoming webhook events
│   ├── workflow.py               # PR review workflow runs
│   └── oauth_state.py            # CSRF state tokens
├── api/v1/
│   ├── webhooks.py               # GitHub webhook receiver
│   ├── repos.py                  # Connect / disconnect / rescan
│   ├── reviews.py                # PR review history
│   ├── events.py                 # Event queries
│   ├── analytics.py              # Team metrics
│   └── memory.py                 # Mem0 CRUD + project map
├── auth/
│   ├── router.py                 # GitHub OAuth flow
│   ├── dependencies.py           # get_current_user() dependency
│   ├── jwt_utils.py              # JWT sign + verify
│   └── token_encryption.py       # Fernet encrypt/decrypt GitHub tokens
├── services/
│   ├── pr_review_service.py      # PR review orchestrator
│   ├── ai_service.py             # Claude integration + parallel agents
│   ├── context_service.py        # Mem0 + Neo4j + MCP context builder
│   ├── graph_builder.py          # Neo4j read + write operations
│   ├── memory_adapter.py         # Mem0 async wrapper
│   ├── memory_extractor.py       # Post-review memory extraction
│   └── project_scanner.py        # Initial repo scan on connect
├── integrations/github/
│   ├── client.py                 # GitHub REST API (diff, files, post review)
│   └── webhook_manager.py        # Install / remove webhooks
└── mcp/
    ├── server.py                 # FastMCP server (4 tools + 1 resource)
    └── client.py                 # MCP client (Linear, Sentry, Slack)

Application Lifecycle

The lifespan context manager in app/main.py handles startup and shutdown:

Startup Sequence

1. Run Alembic migrations - Apply any pending database schema changes
2. Create PostgreSQL tables - Ensure all SQLAlchemy models are present
3. Initialize Neo4j driver - Connect to knowledge graph
4. Create Neo4j schema - Set up constraints and indexes
5. Backfill repos - Scan connected repos not yet in Neo4j (background task)

Shutdown Sequence

1. Dispose PostgreSQL engine - Close all database connections
2. Close Neo4j driver - Gracefully disconnect from graph

# app/main.py:88
@asynccontextmanager
async def lifespan(app: FastAPI):
    # Startup
    await asyncio.to_thread(_run_migrations)  # Alembic
    async with engine.begin() as conn:
        await conn.run_sync(Base.metadata.create_all)
    await init_driver()  # Neo4j
    await create_schema()  # Neo4j constraints
    asyncio.create_task(_scan_unindexed_repos())  # Background backfill
    
    yield
    
    # Shutdown
    await engine.dispose()
    await close_driver()

API Routes

Authentication Routes

GET /auth/github

Starts GitHub OAuth flow. Redirects to github.com/login/oauth/authorize with client_id and scope=repo,read:org.

GET /auth/github/callback

OAuth callback endpoint. Exchanges authorization code for access token, creates/updates user, sets JWT cookie, redirects to frontend.

GET /auth/me

Returns current user profile (requires JWT cookie).

POST /auth/logout

Clears auth cookie.

Webhook Routes

POST /api/v1/webhooks/github

Per-repo webhook receiver

1. Verify HMAC-SHA256 signature against webhook_secret
2. Deduplicate (ignore duplicate events within 1 hour)
3. Create Event row with status=pending
4. Return HTTP 200 immediately (< 1 second)
5. Process PR in background via BackgroundTask

Repository Routes

GET /api/v1/repos

Lists all GitHub repos accessible to the user with connection status.

POST /api/v1/repos/{owner}/{repo}/install

Connects a repository:

1. Create Installation record
2. Install webhook (per-repo secret)
3. Scan file tree → Neo4j graph (background task)

POST /api/v1/repos/{owner}/{repo}/rescan

Re-scans repository file tree and rebuilds Neo4j graph. Useful after repo restructure.

DELETE /api/v1/repos/{owner}/{repo}/install

Disconnects repository:

1. Remove GitHub webhook
2. Mark Installation as inactive
3. (Optional) Delete Neo4j nodes/edges

Review Routes

GET /api/v1/reviews

Returns PR review history with filters:

• repo: Filter by repository
• status: Filter by status (completed, failed, pending)
• limit: Max results (default 20)

Analytics Routes

GET /api/v1/analytics

Returns team metrics:

• Total PRs reviewed
• Verdict distribution (approve/request_changes/needs_discussion)
• Top contributors
• Recent review timeline

Memory Routes

GET /api/v1/memory

Lists Mem0 memories for a repository.

POST /api/v1/memory

Manually adds a project rule or developer pattern to Mem0.

DELETE /api/v1/memory/{id}

Removes a memory from Mem0.

GET /api/v1/memory/project-map

Returns aggregated project context summary (all memories).

Middleware Stack

CORS Middleware

# app/main.py:143
ALLOWED_ORIGINS = [
    "http://localhost:5173",
    "http://localhost:3000",
    "http://localhost:3001",
    settings.FRONTEND_URL,
]

app.add_middleware(
    CORSMiddleware,
    allow_origins=ALLOWED_ORIGINS,
    allow_credentials=True,  # Required for JWT cookies
    allow_methods=["GET", "POST", "DELETE", "PATCH", "OPTIONS"],
    allow_headers=["Content-Type", "Authorization"],
)

allow_credentials=True is required because the frontend sends JWT cookies with withCredentials: true in axios.

Request Logging Middleware

# app/main.py:180
@app.middleware("http")
async def log_requests(request: Request, call_next):
    start = time.time()
    response = await call_next(request)
    duration = round((time.time() - start) * 1000, 2)
    logger.info(f"{request.method} {request.url.path} → {response.status_code} ({duration}ms)")
    return response

Logs every HTTP request with method, path, status code, and duration in milliseconds.

MCP Server Mount

# app/main.py:164
try:
    from app.mcp.server import mcp as _mcp_server
    app.mount("/mcp", _mcp_server.sse_app())
    logger.info("Nectr MCP server mounted at /mcp")
except Exception as _mcp_err:
    logger.warning("MCP server not mounted: %s", _mcp_err)

Why mount() instead of include_router()?

FastMCP’s sse_app() returns a Starlette ASGI application, not an APIRouter. Mounting allows the MCP server to handle its own routing for SSE (GET /mcp/sse) and JSON-RPC (POST /mcp/messages).

Health Check Endpoint

# app/main.py:190
@app.get("/health")
async def health_check():
    uptime_seconds = round(time.time() - startup_time, 1)
    
    # Check PostgreSQL
    db_status = "healthy"
    try:
        async with async_session() as session:
            await session.execute(text("SELECT 1"))
    except Exception as e:
        db_status = f"unhealthy: {str(e)}"
    
    # Check Neo4j
    neo4j_status = "not configured"
    if neo4j_available():
        try:
            async with get_session() as session:
                await session.run("RETURN 1")
            neo4j_status = "healthy"
        except Exception as e:
            neo4j_status = f"unhealthy: {str(e)}"
    
    return {
        "status": "healthy",
        "service": settings.APP_NAME,
        "version": settings.APP_VERSION,
        "environment": settings.APP_ENV,
        "uptime_seconds": uptime_seconds,
        "database": db_status,
        "neo4j": neo4j_status,
    }

Example Response

{
  "status": "healthy",
  "service": "Nectr",
  "version": "1.0.0",
  "environment": "production",
  "uptime_seconds": 3642.8,
  "database": "healthy",
  "neo4j": "healthy"
}

Environment Configuration

All settings are managed via Pydantic in app/core/config.py:

from pydantic_settings import BaseSettings

class Settings(BaseSettings):
    # App
    APP_NAME: str = "Nectr"
    APP_VERSION: str = "1.0.0"
    APP_ENV: str = "development"
    DEBUG: bool = False
    LOG_LEVEL: str = "INFO"
    
    # URLs
    BACKEND_URL: str
    FRONTEND_URL: str
    
    # Database
    DATABASE_URL: str
    
    # Neo4j
    NEO4J_URI: str
    NEO4J_USERNAME: str
    NEO4J_PASSWORD: str
    
    # Auth
    SECRET_KEY: str  # For JWT + Fernet
    GITHUB_CLIENT_ID: str
    GITHUB_CLIENT_SECRET: str
    GITHUB_PAT: str  # Personal access token for posting reviews
    
    # AI
    ANTHROPIC_API_KEY: str
    ANTHROPIC_MODEL: str = "claude-sonnet-4-20250514"
    
    # Mem0
    MEM0_API_KEY: str
    
    # MCP (optional)
    LINEAR_MCP_URL: str | None = None
    LINEAR_API_KEY: str | None = None
    SENTRY_MCP_URL: str | None = None
    SENTRY_AUTH_TOKEN: str | None = None
    SLACK_MCP_URL: str | None = None
    
    # Feature flags
    PARALLEL_REVIEW_AGENTS: bool = False
    
    class Config:
        env_file = ".env"

All MCP integration URLs are optional. If not set, Nectr gracefully skips that integration and logs an info message.

Next Steps

Service Layer

Deep dive into PR review, AI, and context services

Data Flow

Follow a webhook event through the entire system

Database Schema

PostgreSQL tables and relationships

Neo4j Graph

Knowledge graph schema and queries