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 suiteQuick 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 singletonStructured 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 contextCustom 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 validateReview 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.