Server Skills

Server Skills for LlamaFarm

Framework-specific patterns and code review checklists for the LlamaFarm Server component.

Overview

Links to Shared Skills

This skill extends the shared Python skills. See:

• Python Patterns - Dataclasses, comprehensions, imports
• Async Patterns - async/await, asyncio, concurrency
• Typing Patterns - Type hints, generics, Pydantic
• Testing Patterns - Pytest, fixtures, mocking
• Error Handling - Exceptions, logging, context managers
• Security Patterns - Path traversal, injection, secrets

Server-Specific Checklists

Architecture Overview

server/
├── main.py                 # Uvicorn entry point, MCP mount
├── api/
│   ├── main.py             # FastAPI app factory, middleware setup
│   ├── errors.py           # Custom exceptions + exception handlers
│   ├── middleware/         # ASGI middleware (structlog, errors)
│   └── routers/            # API route modules
│       ├── projects/       # Project CRUD endpoints
│       ├── datasets/       # Dataset management
│       ├── rag/            # RAG query endpoints
│       └── ...
├── core/
│   ├── settings.py         # pydantic-settings configuration
│   ├── logging.py          # structlog setup, FastAPIStructLogger
│   └── celery/             # Celery app configuration
│       ├── celery.py       # Celery app instance
│       └── rag_client.py   # RAG task signatures and helpers
├── services/               # Business logic layer
│   ├── project_service.py  # Project CRUD operations
│   ├── dataset_service.py  # Dataset management
│   └── ...
├── agents/                 # AI agent implementations
└── tests/                  # Pytest test suite

Quick Reference

Settings Pattern (pydantic-settings)

from pydantic_settings import BaseSettings

class Settings(BaseSettings, env_file=".env"):
    HOST: str = "0.0.0.0"
    PORT: int = 14345
    LOG_LEVEL: str = "INFO"

settings = Settings()  # Module-level singleton

Structured Logging

from core.logging import FastAPIStructLogger

logger = FastAPIStructLogger(__name__)
logger.info("Operation completed", extra={"count": 10, "duration_ms": 150})
logger.bind(namespace=namespace, project=project_id)  # Add context

Custom Exceptions

# Define exception hierarchy
class NotFoundError(Exception): ...
class ProjectNotFoundError(NotFoundError):
    def __init__(self, namespace: str, project_id: str):
        self.namespace = namespace
        self.project_id = project_id
        super().__init__(f"Project {namespace}/{project_id} not found")

# Register handler in api/errors.py
async def _handle_project_not_found(request: Request, exc: Exception) -> Response:
    payload = ErrorResponse(error="ProjectNotFound", message=str(exc))
    return JSONResponse(status_code=404, content=payload.model_dump())

def register_exception_handlers(app: FastAPI) -> None:
    app.add_exception_handler(ProjectNotFoundError, _handle_project_not_found)

Service Layer Pattern

class ProjectService:
    @classmethod
    def get_project(cls, namespace: str, project_id: str) -> Project:
        project_dir = cls.get_project_dir(namespace, project_id)
        if not os.path.isdir(project_dir):
            raise ProjectNotFoundError(namespace, project_id)
        # ... load and validate

Review Checklist Summary

1. FastAPI Routes (High priority)

• Proper async/sync function choice
• Response model defined with response_model=
• OpenAPI metadata (operation_id, tags, summary)
• HTTPException with proper status codes

2. Celery Tasks (High priority)

• Use signatures for cross-service calls
• Implement proper timeout and polling
• Handle task failures gracefully
• Store group metadata for parallel tasks

3. Pydantic Models (Medium priority)

• Use Pydantic v2 patterns (model_config, Field)
• Proper validation with field constraints
• Serialization with model_dump()

4. Performance (Medium priority)

• Avoid blocking calls in async functions
• Use proper connection pooling for external services
• Implement caching where appropriate

See individual topic files for detailed checklists with grep patterns.

Celery Patterns for LlamaFarm Server

Best practices and code review checklist for Celery task patterns, error handling, and cross-service communication.

Architecture Overview

The Server uses Celery to dispatch tasks to the RAG worker. Key concepts:

• Signatures - Build task calls without executing them
• Groups - Parallel execution of multiple tasks
• Polling - Server polls for task completion (no direct result access)
• Filesystem Broker - Default broker for local development

Server (FastAPI) --> Celery Signature --> Filesystem Broker --> RAG Worker
                                                            <-- Result Backend

---

Ideal Patterns

Building Task Signatures

from celery import signature
from core.celery import app

def build_ingest_signature(
    project_dir: str,
    database_name: str,
    source_path: str,
) -> signature:
    """Build a Celery signature for the rag.ingest_file task."""
    return signature(
        "rag.ingest_file",
        args=[project_dir, database_name, source_path],
        app=app,
    )

Async Polling for Task Completion

import asyncio
from typing import Any

async def ingest_file_with_rag(
    project_dir: str,
    database_name: str,
    source_path: str,
    timeout: int = 300,
    poll_interval: float = 2.0,
) -> tuple[bool, dict[str, Any]]:
    """Dispatch task and poll for completion asynchronously."""
    sig = build_ingest_signature(project_dir, database_name, source_path)
    result = sig.apply_async()

    waited = 0.0
    while waited < timeout:
        try:
            status = result.status
            if status not in ("PENDING", "STARTED"):
                break
        except Exception:
            await asyncio.sleep(poll_interval)
            waited += poll_interval
            continue

        await asyncio.sleep(poll_interval)
        waited += poll_interval

    if result.status == "SUCCESS":
        return True, result.result
    elif result.status == "FAILURE":
        if hasattr(result, "traceback") and result.traceback:
            raise Exception(f"Task failed: {result.traceback}")
        raise Exception("Task failed")

    return False, {"error": f"Task timed out with status: {result.status}"}

Synchronous Polling Helper

import time

def _run_sync_task_with_polling(
    task_signature,
    timeout: float,
    poll_interval: float
) -> Any:
    """Helper for synchronous contexts to poll a Celery AsyncResult."""
    result = task_signature.apply_async()
    waited = 0.0

    while waited < timeout:
        try:
            status = result.status
            if status not in ("PENDING", "STARTED"):
                break
        except Exception:
            time.sleep(poll_interval)
            waited += poll_interval
            continue
        time.sleep(poll_interval)
        waited += poll_interval

    if result.status == "SUCCESS":
        return result.result
    elif result.status == "FAILURE":
        if hasattr(result, "traceback") and result.traceback:
            raise Exception(f"Task failed: {result.traceback}")
        raise Exception("Task failed")

    return None

Group Task with Metadata Storage

from celery import group
from core.celery import app

def dispatch_parallel_ingestion(
    namespace: str,
    project_id: str,
    file_paths: list[str],
) -> str:
    """Dispatch multiple files for parallel processing."""
    # Build signatures for each file
    signatures = [
        build_ingest_signature(project_dir, database, path)
        for path in file_paths
    ]

    # Create and dispatch group
    task_group = group(signatures)
    group_result = task_group.apply_async()

    # Store metadata for status tracking
    # (GroupResult.restore() doesn't work well with filesystem backend)
    task_id = str(uuid.uuid4())
    metadata = {
        "type": "group",
        "namespace": namespace,
        "project": project_id,
        "children": [child.id for child in group_result.results],
        "file_hashes": [hash_file(p) for p in file_paths],
    }
    app.backend.store_result(task_id, metadata, "PENDING")

    return task_id

---

Checklist

1. Use Signatures for Cross-Service Calls

Description: Never import tasks directly from other services. Use signatures.

Search Pattern:

grep -rn "from rag\." server/ | grep -v "\.pyc"

Pass Criteria: No direct imports from rag package. Use signature("rag.task_name").

Severity: High

Recommendation: Use signature pattern:

task = signature("rag.ingest_file", args=[...], app=app)
result = task.apply_async()

---

2. Async Polling Uses asyncio.sleep

Description: Async functions must use asyncio.sleep(), not time.sleep().

Search Pattern:

grep -rn "time\.sleep" server/ | grep "async def" -B10 | grep "time\.sleep"

Pass Criteria: No time.sleep() in async functions.

Severity: High

Recommendation: Replace with await asyncio.sleep(interval).

---

3. Task Timeout Implemented

Description: All task polling must have a timeout to prevent infinite waits.

Search Pattern:

grep -rn "while.*result\." server/core/celery/ | grep -v "timeout\|waited"

Pass Criteria: Every polling loop has a timeout condition.

Severity: High

Recommendation: Implement timeout pattern:

waited = 0.0
while waited < timeout:
    # ... poll logic
    waited += poll_interval

---

4. Task Failure Handling

Description: Handle task failures gracefully with proper error messages.

Search Pattern:

grep -rn "result\.status" server/ | grep -v "FAILURE"

Pass Criteria: Every status check handles FAILURE state.

Severity: High

Recommendation:

if result.status == "FAILURE":
    if hasattr(result, "traceback") and result.traceback:
        raise Exception(f"Task failed: {result.traceback}")
    raise Exception("Task failed")

---

5. Group Metadata Stored for Tracking

Description: Parallel tasks need metadata storage for status tracking.

Search Pattern:

grep -rn "group(" server/ | grep -v "store_result"

Pass Criteria: Group tasks store metadata with children IDs.

Severity: Medium

Recommendation: Store group metadata:

metadata = {
    "type": "group",
    "children": [child.id for child in group_result.results],
}
app.backend.store_result(task_id, metadata, "PENDING")

---

6. Celery App Configuration Complete

Description: Celery app must be configured with proper serialization and routing.

Search Pattern:

grep -rn "app.conf.update" server/core/celery/celery.py

Pass Criteria: Configuration includes serializer, timezone, and task routes.

Severity: Medium

Recommendation:

app.conf.update({
    "task_serializer": "json",
    "accept_content": ["json"],
    "result_serializer": "json",
    "timezone": "UTC",
    "enable_utc": True,
    "task_routes": {
        "rag.*": {"queue": "rag"},
        "core.celery.tasks.*": {"queue": "server"},
    },
})

---

7. Result Backend Path Handles Windows

Description: File paths for filesystem backend must handle Windows paths.

Search Pattern:

grep -rn "file://" server/core/celery/

Pass Criteria: Windows paths use file:///C:/... format.

Severity: Medium

Recommendation:

result_backend_path = path.replace("\\", "/")
if sys.platform == "win32" and path[1] == ":":
    result_backend_url = f"file:///{result_backend_path}"
else:
    result_backend_url = f"file://{result_backend_path}"

---

8. Prevent Celery Logger Override

Description: Prevent Celery from overriding structlog configuration.

Search Pattern:

grep -rn "setup_logging.connect" server/core/celery/

Pass Criteria: Empty setup_celery_logging signal handler exists.

Severity: Low

Recommendation:

from celery import signals

@signals.setup_logging.connect
def setup_celery_logging(**kwargs):
    pass  # Prevent Celery from overriding root logger

---

9. Task Cancellation Implemented

Description: Long-running tasks should support cancellation.

Search Pattern:

grep -rn "revoke" server/api/routers/

Pass Criteria: Cancellation endpoint uses app.control.revoke().

Severity: Medium

Recommendation:

for child_id in child_task_ids:
    child_result = celery_app.AsyncResult(child_id)
    if child_result.state == "PENDING":
        celery_app.control.revoke(child_id, terminate=False)

---

10. Default Return Values for Timeouts

Description: Provide sensible defaults when tasks timeout.

Search Pattern:

grep -rn "_run_sync_task_with_polling" server/core/celery/rag_client.py -A2 | grep "or {"

Pass Criteria: Functions return default values on timeout, not None.

Severity: Low

Recommendation:

return _run_sync_task_with_polling(task, timeout=30) or {
    "status": "degraded",
    "message": "Task timed out",
}

---

Anti-Patterns to Avoid

1. Direct Task Import

# BAD - creates import dependency
from rag.tasks import ingest_file
ingest_file.delay(...)

# GOOD - loose coupling
task = signature("rag.ingest_file", args=[...], app=app)
task.apply_async()

2. Blocking Sleep in Async

# BAD - blocks event loop
async def poll_task():
    while True:
        time.sleep(1)  # Blocks!

# GOOD - non-blocking
async def poll_task():
    while True:
        await asyncio.sleep(1)

3. Missing Timeout

# BAD - infinite loop possible
while result.status == "PENDING":
    time.sleep(1)

# GOOD - bounded wait
waited = 0.0
while waited < timeout and result.status == "PENDING":
    time.sleep(poll_interval)
    waited += poll_interval

4. Ignoring Task Traceback

# BAD - loses error context
if result.status == "FAILURE":
    raise Exception("Task failed")

# GOOD - includes traceback
if result.status == "FAILURE":
    if hasattr(result, "traceback") and result.traceback:
        raise Exception(f"Task failed: {result.traceback}")
    raise Exception("Task failed")

FastAPI Patterns for LlamaFarm Server

Best practices and code review checklist for FastAPI routes, middleware, and dependencies.

Route Definition Patterns

Ideal Route Structure

from fastapi import APIRouter, HTTPException, Response
from fastapi import Path as FastAPIPath
from pydantic import BaseModel, Field

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

class GetProjectResponse(BaseModel):
    project: Project = Field(..., description="The project details")

@router.get(
    "/{namespace}/{project_id}",
    operation_id="project_get",
    summary="Get a project",
    tags=["projects", "mcp"],
    response_model=GetProjectResponse,
    responses={
        200: {"model": GetProjectResponse},
        404: {"model": ErrorResponse},
    },
)
async def get_project(
    namespace: str = FastAPIPath(..., description="The namespace"),
    project_id: str = FastAPIPath(..., description="The project ID"),
) -> GetProjectResponse:
    """Get a project by namespace and ID."""
    project = ProjectService.get_project(namespace, project_id)
    return GetProjectResponse(project=project)

Key Elements

1. Router with prefix and tags - Groups related endpoints 2. Operation ID - Unique identifier for OpenAPI/code generation 3. Response models - Pydantic models for type safety 4. Path parameter descriptions - Use FastAPIPath for documentation 5. Docstring - Appears in OpenAPI documentation

---

Checklist

1. Route Has Response Model

Description: All routes should define a response model for type safety and documentation.

Search Pattern:

grep -rn "@router\.\(get\|post\|put\|delete\|patch\)" server/api/routers/ | grep -v "response_model"

Pass Criteria: Every route decorator includes response_model= or is documented in responses={}.

Severity: Medium

Recommendation: Add response_model=ResponseClass to route decorators or define response in responses={} dict.

---

2. Route Has Operation ID

Description: Routes exposed via MCP or used for code generation need unique operation_id.

Search Pattern:

grep -rn "@router\.\(get\|post\|put\|delete\)" server/api/routers/ -A5 | grep -E "tags=.*mcp" | grep -v "operation_id"

Pass Criteria: All routes with tags=["mcp"] have an operation_id.

Severity: Medium

Recommendation: Add operation_id="resource_action" (e.g., project_get, dataset_create).

---

3. Async Functions for I/O Operations

Description: Routes performing I/O (database, file, network) should be async.

Search Pattern:

grep -rn "^def " server/api/routers/ | grep -v "__init__" | grep -v "test_"

Pass Criteria: Route handlers performing I/O use async def.

Severity: High

Recommendation: Convert to async def and use await for I/O operations.

---

4. HTTPException with Proper Status Codes

Description: Use HTTPException with appropriate status codes, not bare exceptions.

Search Pattern:

grep -rn "raise Exception" server/api/routers/

Pass Criteria: No bare raise Exception in route handlers. Use HTTPException or custom exceptions.

Severity: High

Recommendation: Replace with raise HTTPException(status_code=4xx/5xx, detail="message").

---

5. Path Traversal Prevention

Description: User-provided paths must be validated to prevent directory traversal.

Search Pattern:

grep -rn "os.path.join.*namespace\|os.path.join.*project" server/ | grep -v "normpath\|resolve"

Pass Criteria: All path constructions validate against base path after normalization.

Severity: Critical

Recommendation: Use pattern:

norm_path = os.path.normpath(raw_path)
if not norm_path.startswith(os.path.abspath(base_path) + os.sep):
    raise NamespaceNotFoundError("Invalid: path traversal detected")

---

6. Exception Handlers Registered

Description: Custom exceptions should have registered handlers for consistent API responses.

Search Pattern:

grep -rn "class.*Error.*Exception" server/api/errors.py

Pass Criteria: Each custom exception has a corresponding handler in register_exception_handlers().

Severity: Medium

Recommendation: Add handler in api/errors.py:

async def _handle_my_error(request: Request, exc: Exception) -> Response:
    payload = ErrorResponse(error="MyError", message=str(exc))
    return JSONResponse(status_code=4xx, content=payload.model_dump())

app.add_exception_handler(MyError, _handle_my_error)

---

7. Response Headers Set Correctly

Description: Custom headers (e.g., session IDs) should be set on Response object.

Search Pattern:

grep -rn "response.headers\[" server/api/routers/

Pass Criteria: Headers set via response.headers["X-Header"] = value.

Severity: Low

Recommendation: Inject response: Response parameter and set headers:

async def endpoint(response: Response):
    response.headers["X-Session-ID"] = session_id

---

8. Middleware Order Correct

Description: Middleware is applied in reverse order - last added runs first on request.

Search Pattern:

grep -rn "add_middleware" server/api/main.py

Pass Criteria: Error handling middleware added after logging middleware.

Severity: Medium

Recommendation: Order in api/main.py:

app.add_middleware(ErrorHandlerMiddleware)  # Added last, runs first
app.add_middleware(StructLogMiddleware)
app.add_middleware(CorrelationIdMiddleware)  # Added first, runs last

---

9. Lifespan Context Manager Used

Description: Use lifespan for startup/shutdown instead of deprecated events.

Search Pattern:

grep -rn "@app.on_event" server/

Pass Criteria: No @app.on_event("startup") or @app.on_event("shutdown").

Severity: Medium

Recommendation: Use lifespan context manager:

@asynccontextmanager
async def lifespan(app: FastAPI):
    # Startup
    logger.info("Starting API")
    yield
    # Shutdown
    await cleanup_resources()

app = FastAPI(lifespan=lifespan)

---

10. Streaming Responses Handled Correctly

Description: SSE/streaming responses should use StreamingResponse with proper content type.

Search Pattern:

grep -rn "StreamingResponse\|EventSourceResponse" server/api/routers/

Pass Criteria: Streaming endpoints return proper response type with correct media_type.

Severity: Medium

Recommendation:

from starlette.responses import StreamingResponse

return StreamingResponse(
    generator(),
    media_type="text/event-stream",
    headers={"Cache-Control": "no-cache"}
)

---

Anti-Patterns to Avoid

1. Blocking Calls in Async Routes

# BAD - blocks event loop
@router.get("/data")
async def get_data():
    time.sleep(5)  # Blocks!
    data = requests.get(url)  # Blocks!

# GOOD - non-blocking
@router.get("/data")
async def get_data():
    await asyncio.sleep(5)
    async with httpx.AsyncClient() as client:
        data = await client.get(url)

2. Missing Error Context

# BAD - loses original error
except Exception:
    raise HTTPException(status_code=500)

# GOOD - chains exceptions
except Exception as e:
    raise HTTPException(status_code=500, detail=str(e)) from e

3. Hardcoded Status Codes Without Model

# BAD - no type safety
return {"status": "ok"}

# GOOD - typed response
class StatusResponse(BaseModel):
    status: str

@router.get("/", response_model=StatusResponse)
async def health():
    return StatusResponse(status="ok")

Performance Patterns for LlamaFarm Server

Best practices and code review checklist for server-specific optimizations, async patterns, and resource management.

Overview

Performance considerations for the LlamaFarm Server:

• Async I/O - Non-blocking operations for HTTP, file, and database access
• Connection Pooling - Reuse HTTP clients and database connections
• Caching - In-memory caching for frequently accessed data
• Resource Cleanup - Proper lifecycle management with context managers

---

Ideal Patterns

HTTP Client Reuse

import httpx
from contextlib import asynccontextmanager

# Module-level client for reuse
_http_client: httpx.AsyncClient | None = None

async def get_http_client() -> httpx.AsyncClient:
    """Get or create a shared HTTP client."""
    global _http_client
    if _http_client is None:
        _http_client = httpx.AsyncClient(
            timeout=30.0,
            limits=httpx.Limits(max_connections=100, max_keepalive_connections=20)
        )
    return _http_client

async def cleanup_http_client():
    """Cleanup on shutdown."""
    global _http_client
    if _http_client:
        await _http_client.aclose()
        _http_client = None

Async File Operations

import aiofiles
from pathlib import Path

async def read_file_async(path: Path) -> str:
    """Read file without blocking the event loop."""
    async with aiofiles.open(path, mode='r') as f:
        return await f.read()

async def write_file_async(path: Path, content: str) -> None:
    """Write file without blocking the event loop."""
    async with aiofiles.open(path, mode='w') as f:
        await f.write(content)

In-Memory Session Cache with TTL

import threading
import time
from dataclasses import dataclass

SESSION_TTL_SECONDS = 30 * 60  # 30 minutes

@dataclass
class SessionRecord:
    agent: ChatOrchestratorAgent
    created_at: float
    last_used: float
    request_count: int

agent_sessions: dict[str, SessionRecord] = {}
_agent_sessions_lock = threading.RLock()

def _cleanup_expired_sessions(now: float | None = None) -> None:
    """Remove expired sessions from cache."""
    timestamp = now or time.time()
    to_delete = [
        key for key, record in agent_sessions.items()
        if timestamp - record.last_used > SESSION_TTL_SECONDS
    ]
    for key in to_delete:
        agent_sessions.pop(key, None)

Streaming Response for Large Data

from starlette.responses import StreamingResponse
from typing import AsyncIterator

async def stream_chat_response(
    messages: list,
) -> AsyncIterator[str]:
    """Stream chat responses as SSE events."""
    async for chunk in agent.stream_chat(messages):
        yield f"data: {chunk.model_dump_json()}\n\n"
    yield "data: [DONE]\n\n"

@router.post("/chat/stream")
async def chat_stream(request: ChatRequest):
    return StreamingResponse(
        stream_chat_response(request.messages),
        media_type="text/event-stream",
        headers={
            "Cache-Control": "no-cache",
            "Connection": "keep-alive",
        }
    )

Lifespan for Resource Management

from contextlib import asynccontextmanager
from fastapi import FastAPI

@asynccontextmanager
async def lifespan(app: FastAPI):
    """Manage application lifecycle resources."""
    # Startup
    logger.info("Starting LlamaFarm API")
    await initialize_connection_pools()
    yield
    # Shutdown
    logger.info("Shutting down LlamaFarm API")
    await cleanup_all_mcp_services()
    await cleanup_http_client()
    logger.info("Shutdown complete")

app = FastAPI(lifespan=lifespan)

---

Checklist

1. No Blocking Calls in Async Functions

Description: Async route handlers must not use blocking I/O.

Search Pattern:

grep -rn "async def" server/api/routers/ -A20 | grep -E "time\.sleep|requests\.|open\(|os\.read"

Pass Criteria: No time.sleep(), requests.*, or synchronous file I/O in async functions.

Severity: Critical

Recommendation:

# BAD
async def handler():
    time.sleep(1)  # Blocks!
    data = requests.get(url)  # Blocks!

# GOOD
async def handler():
    await asyncio.sleep(1)
    async with httpx.AsyncClient() as client:
        data = await client.get(url)

---

2. HTTP Client Reused

Description: Create HTTP clients once and reuse them across requests.

Search Pattern:

grep -rn "httpx\.\(Client\|AsyncClient\)()" server/ | grep -v "_client\|global"

Pass Criteria: No inline client creation per request.

Severity: Medium

Recommendation: Use module-level client or dependency injection:

# Module level
_client = httpx.AsyncClient(timeout=30.0)

# Or as dependency
async def get_client() -> httpx.AsyncClient:
    return shared_client

---

3. Session Cache Has TTL Cleanup

Description: In-memory caches must implement TTL-based cleanup.

Search Pattern:

grep -rn "dict\[str.*\].*=" server/ | grep -v "cleanup\|TTL\|expire"

Pass Criteria: Dict-based caches have cleanup mechanism.

Severity: Medium

Recommendation: Implement periodic cleanup:

def _cleanup_expired_sessions(now: float | None = None) -> None:
    timestamp = now or time.time()
    to_delete = [
        key for key, record in sessions.items()
        if timestamp - record.last_used > TTL_SECONDS
    ]
    for key in to_delete:
        sessions.pop(key, None)

---

4. Thread Lock for Shared State

Description: Shared mutable state requires thread synchronization.

Search Pattern:

grep -rn "^[a-z_]*_sessions\|^[a-z_]*_cache" server/ | grep "dict\|{}"

Pass Criteria: Module-level mutable state has associated lock.

Severity: High

Recommendation:

agent_sessions: dict[str, SessionRecord] = {}
_agent_sessions_lock = threading.RLock()

with _agent_sessions_lock:
    agent_sessions[key] = record

---

5. Streaming for Large Responses

Description: Large responses should use streaming to reduce memory.

Search Pattern:

grep -rn "return.*\[.*for.*in" server/api/routers/ | grep -v "StreamingResponse"

Pass Criteria: List comprehensions for large data use streaming.

Severity: Medium

Recommendation:

# BAD - loads all into memory
return [process(item) for item in large_list]

# GOOD - streams results
async def stream_results():
    for item in large_list:
        yield process(item)
return StreamingResponse(stream_results())

---

6. Context Managers for Resources

Description: External resources should use context managers for cleanup.

Search Pattern:

grep -rn "open(" server/ | grep -v "with\|async with" | grep -v ".venv"

Pass Criteria: All open() calls use with or async with.

Severity: Medium

Recommendation:

# BAD - may leak file handle
f = open(path)
data = f.read()

# GOOD - automatic cleanup
with open(path) as f:
    data = f.read()

---

7. Lifespan Used for Startup/Shutdown

Description: Use lifespan context manager instead of deprecated events.

Search Pattern:

grep -rn "on_event" server/

Pass Criteria: No @app.on_event("startup") or @app.on_event("shutdown").

Severity: Medium

Recommendation:

@asynccontextmanager
async def lifespan(app: FastAPI):
    # Startup
    await setup()
    yield
    # Shutdown
    await cleanup()

app = FastAPI(lifespan=lifespan)

---

8. Efficient JSON Serialization

Description: Use efficient JSON serialization for responses.

Search Pattern:

grep -rn "json\.dumps" server/api/ | grep -v "orjson\|ujson"

Pass Criteria: Standard json.dumps used appropriately or faster alternatives for hot paths.

Severity: Low

Recommendation: For high-throughput endpoints, consider orjson:

import orjson
from fastapi.responses import ORJSONResponse

app = FastAPI(default_response_class=ORJSONResponse)

---

9. Pagination for List Endpoints

Description: List endpoints should support pagination for large datasets.

Search Pattern:

grep -rn "def list_\|async def list_" server/api/routers/ | grep -v "limit\|offset\|page"

Pass Criteria: List endpoints accept pagination parameters.

Severity: Medium

Recommendation:

@router.get("/items")
async def list_items(
    limit: int = Query(100, ge=1, le=1000),
    offset: int = Query(0, ge=0),
) -> ListResponse:
    items = service.list_items(limit=limit, offset=offset)
    return ListResponse(items=items, total=total)

---

10. Logging Level Checks

Description: Expensive log message construction should check level first.

Search Pattern:

grep -rn "logger\.debug(" server/ | grep -E "\+|format|%|f\""

Pass Criteria: Debug logs with expensive formatting check level or use lazy evaluation.

Severity: Low

Recommendation:

# BAD - always constructs string
logger.debug(f"Processing {expensive_operation()}")

# GOOD - structlog handles this efficiently
logger.debug("Processing", result=value)

# Or check level
if logger.isEnabledFor(logging.DEBUG):
    logger.debug(f"Processing {expensive_operation()}")

---

Anti-Patterns to Avoid

1. Creating Clients Per Request

# BAD - creates new connection each time
async def handler():
    async with httpx.AsyncClient() as client:
        return await client.get(url)

# GOOD - reuse client
async def handler():
    client = await get_shared_client()
    return await client.get(url)

2. Unbounded Cache Growth

# BAD - never cleans up
cache = {}
def get_cached(key):
    if key not in cache:
        cache[key] = compute(key)
    return cache[key]

# GOOD - with TTL cleanup
def get_cached(key):
    cleanup_expired()
    if key not in cache or is_expired(cache[key]):
        cache[key] = CacheEntry(value=compute(key), expires=time.time() + TTL)
    return cache[key].value

3. Blocking File I/O in Async

# BAD - blocks event loop
async def read_config():
    with open("config.yaml") as f:
        return yaml.safe_load(f)

# GOOD - use aiofiles or run_in_executor
async def read_config():
    async with aiofiles.open("config.yaml") as f:
        content = await f.read()
    return yaml.safe_load(content)

4. Missing Response Streaming

# BAD - loads all in memory
@router.get("/export")
async def export_data():
    data = await get_all_records()  # Could be millions
    return {"records": data}

# GOOD - stream response
@router.get("/export")
async def export_data():
    async def generate():
        async for record in get_records_stream():
            yield record.model_dump_json() + "\n"
    return StreamingResponse(generate(), media_type="application/x-ndjson")

Pydantic Patterns for LlamaFarm Server

Best practices and code review checklist for Pydantic v2 models, validation, and serialization.

Overview

LlamaFarm uses Pydantic v2 throughout:

• API Models - Request/response schemas in route handlers
• Settings - Configuration via pydantic-settings
• Domain Models - Internal data structures with validation
• Config Models - LlamaFarmConfig from the config package

---

Ideal Patterns

API Request/Response Models

from pydantic import BaseModel, Field, ConfigDict

class CreateProjectRequest(BaseModel):
    """Request model for creating a new project."""
    name: str = Field(..., description="The name of the project")
    config_template: str | None = Field(
        None,
        description="The config template to use for the project"
    )

class CreateProjectResponse(BaseModel):
    """Response model for project creation."""
    project: Project = Field(..., description="The created project")

    model_config = ConfigDict(
        str_strip_whitespace=True,
        json_schema_extra={"example": {"project": {"name": "my-project"}}}
    )

Settings with pydantic-settings

from pydantic_settings import BaseSettings
from pathlib import Path

default_data_dir = str(Path.home() / ".llamafarm")

class Settings(BaseSettings, env_file=".env"):
    """Application settings loaded from environment."""
    HOST: str = "0.0.0.0"
    PORT: int = 14345
    LOG_LEVEL: str = "INFO"
    LOG_JSON_FORMAT: bool = False

    lf_data_dir: str = default_data_dir
    celery_broker_url: str = ""

settings = Settings()  # Module-level singleton

Model with Validation

from pydantic import BaseModel, Field, field_validator
from datetime import datetime

class Project(BaseModel):
    """Project domain model with validation."""
    namespace: str
    name: str
    config: LlamaFarmConfig
    validation_error: str | None = None
    last_modified: datetime | None = None

    @field_validator("name")
    @classmethod
    def name_not_empty(cls, v: str) -> str:
        if not v or not v.strip():
            raise ValueError("name cannot be empty")
        return v.strip()

Union Types for Flexible Responses

from pydantic import BaseModel
from config.datamodel import LlamaFarmConfig

class Project(BaseModel):
    """Project that can hold validated config or raw dict."""
    namespace: str
    name: str
    config: LlamaFarmConfig | dict = Field(
        ...,
        description="The configuration (validated model or raw dict)"
    )
    validation_error: str | None = Field(
        None,
        description="Validation error message if config has issues"
    )

Serialization with model_dump

# Serialize to dict
config_dict = config.model_dump(mode="json", exclude_none=True)

# Serialize for JSON response
response_data = model.model_dump(mode="json", exclude_none=True)

# Serialize specific fields
partial = model.model_dump(include={"name", "namespace"})

---

Checklist

1. Use Pydantic v2 Patterns

Description: Use Pydantic v2 APIs, not deprecated v1 patterns.

Search Pattern:

grep -rn "class Config:" server/ | grep -v ".venv"

Pass Criteria: No class Config: inside models. Use model_config = ConfigDict(...).

Severity: Medium

Recommendation: Migrate to v2:

# v1 (deprecated)
class MyModel(BaseModel):
    class Config:
        str_strip_whitespace = True

# v2 (correct)
class MyModel(BaseModel):
    model_config = ConfigDict(str_strip_whitespace=True)

---

2. Field Descriptions Provided

Description: API-facing models should have Field descriptions for OpenAPI docs.

Search Pattern:

grep -rn "Field(\.\.\." server/api/routers/ | grep -v "description="

Pass Criteria: Required fields have descriptions.

Severity: Low

Recommendation: Add descriptions:

name: str = Field(..., description="The name of the resource")

---

3. Use model_dump Not dict()

Description: Use model_dump() instead of deprecated .dict() method.

Search Pattern:

grep -rn "\.dict()" server/ | grep -v ".venv"

Pass Criteria: No .dict() calls on Pydantic models.

Severity: Medium

Recommendation: Replace with model_dump():

# v1 (deprecated)
data = model.dict(exclude_none=True)

# v2 (correct)
data = model.dump(mode="json", exclude_none=True)

---

4. Use model_validate Not parse_obj

Description: Use model_validate() instead of deprecated parse_obj().

Search Pattern:

grep -rn "parse_obj\|parse_raw" server/ | grep -v ".venv"

Pass Criteria: No parse_obj() or parse_raw() calls.

Severity: Medium

Recommendation: Replace with v2 methods:

# v1 (deprecated)
model = MyModel.parse_obj(data)

# v2 (correct)
model = MyModel.model_validate(data)

---

5. Validators Use @field_validator

Description: Use @field_validator decorator, not deprecated @validator.

Search Pattern:

grep -rn "@validator" server/ | grep -v ".venv"

Pass Criteria: No @validator decorators. Use @field_validator.

Severity: Medium

Recommendation:

# v1 (deprecated)
@validator("name")
def validate_name(cls, v):
    return v.strip()

# v2 (correct)
@field_validator("name")
@classmethod
def validate_name(cls, v: str) -> str:
    return v.strip()

---

6. Model Serializers Use @model_serializer

Description: Use @model_serializer for custom serialization, not __json__.

Search Pattern:

grep -rn "def __json__" server/ | grep -v ".venv"

Pass Criteria: No __json__ methods. Use @model_serializer.

Severity: Low

Recommendation:

from pydantic import model_serializer

class MyModel(BaseModel):
    @model_serializer
    def serialize(self) -> dict:
        return {"custom": "serialization"}

---

7. Proper Optional Type Hints

Description: Use T | None syntax for optional fields, not Optional[T].

Search Pattern:

grep -rn "Optional\[" server/ | grep -v ".venv" | grep -v "typing import"

Pass Criteria: Use modern T | None syntax.

Severity: Low

Recommendation:

# Old style
from typing import Optional
field: Optional[str] = None

# Modern style (Python 3.10+)
field: str | None = None

---

8. ConfigDict Used for Model Configuration

Description: Use ConfigDict for model configuration, not class Config.

Search Pattern:

grep -rn "model_config\s*=" server/api/ | grep -v "ConfigDict"

Pass Criteria: model_config assigned from ConfigDict().

Severity: Low

Recommendation:

from pydantic import ConfigDict

class MyModel(BaseModel):
    model_config = ConfigDict(
        str_strip_whitespace=True,
        use_enum_values=True,
    )

---

9. Exclude None in JSON Responses

Description: API responses should exclude None values for cleaner JSON.

Search Pattern:

grep -rn "model_dump(" server/api/ | grep -v "exclude_none"

Pass Criteria: API serialization uses exclude_none=True.

Severity: Low

Recommendation:

# In response
return model.model_dump(mode="json", exclude_none=True)

# Or use custom response class
class NoNoneJSONResponse(JSONResponse):
    def render(self, content) -> bytes:
        return super().render(jsonable_encoder(content, exclude_none=True))

---

10. Proper Error Extraction from ValidationError

Description: Extract structured errors from Pydantic ValidationError.

Search Pattern:

grep -rn "ValidationError" server/ | grep -v "import\|.venv"

Pass Criteria: ValidationError handling extracts structured error messages.

Severity: Medium

Recommendation:

except ValidationError as e:
    if hasattr(e, "errors") and callable(e.errors):
        error_details = []
        for err in e.errors():
            loc = ".".join(str(x) for x in err.get("loc", []))
            msg = err.get("msg", "validation error")
            error_details.append(f"{loc}: {msg}")
        validation_error_msg = "; ".join(error_details[:5])

---

Anti-Patterns to Avoid

1. Mutable Default Arguments

# BAD - mutable default
class Config(BaseModel):
    items: list = []

# GOOD - use Field with default_factory
class Config(BaseModel):
    items: list = Field(default_factory=list)

2. Mixing v1 and v2 APIs

# BAD - mixed APIs
class MyModel(BaseModel):
    class Config:  # v1
        ...
    model_config = ConfigDict(...)  # v2

# GOOD - v2 only
class MyModel(BaseModel):
    model_config = ConfigDict(...)

3. Bare dict Instead of TypedDict or Model

# BAD - no type safety
def process(data: dict) -> dict:
    return {"result": data["value"]}

# GOOD - typed models
class InputData(BaseModel):
    value: str

class OutputData(BaseModel):
    result: str

def process(data: InputData) -> OutputData:
    return OutputData(result=data.value)

4. Missing Field Constraints

# BAD - no constraints
class User(BaseModel):
    age: int
    email: str

# GOOD - with constraints
class User(BaseModel):
    age: int = Field(..., ge=0, le=150)
    email: str = Field(..., pattern=r"^[\w\.-]+@[\w\.-]+\.\w+$")