name: server-skills
description: Server-specific best practices for FastAPI, Celery, and Pydantic. Extends python-skills with framework-specific patterns.
allowed-tools: Read, Grep, Glob, Bash
user-invocable: false

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 = 8000
    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)
2. Proper async/sync function choice
3. Response model defined with response_model=
4. OpenAPI metadata (operation_id, tags, summary)

5. HTTPException with proper status codes

6. Celery Tasks (High priority)

7. Use signatures for cross-service calls
8. Implement proper timeout and polling
9. Handle task failures gracefully

10. Store group metadata for parallel tasks

11. Pydantic Models (Medium priority)

12. Use Pydantic v2 patterns (model_config, Field)
13. Proper validation with field constraints

14. Serialization with model_dump()

15. Performance (Medium priority)

16. Avoid blocking calls in async functions
17. Use proper connection pooling for external services
18. Implement caching where appropriate

See individual topic files for detailed checklists with grep patterns.