Server Skills for LlamaFarm

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

Overview

Property Value

Path server/

Python 3.12+

Framework FastAPI 0.116+

Task Queue Celery 5.5+

Validation Pydantic 2.x, pydantic-settings

Logging structlog with FastAPIStructLogger

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

Topic File Key Points

FastAPI fastapi.md Routes, dependencies, middleware, exception handlers

Celery celery.md Task patterns, error handling, retries, signatures

Pydantic pydantic.md Pydantic v2 models, validation, serialization

Performance performance.md Async patterns, caching, connection pooling

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

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

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

Pydantic Models (Medium priority)

• Use Pydantic v2 patterns (model_config, Field)

• Proper validation with field constraints

• Serialization with model_dump()

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.