server-skills

LlamaFarm のサーバースキル

LlamaFarm Server コンポーネントに特化したフレームワークのパターンとコードレビュー用チェックリストです。

概要

プロパティ	値
パス	server/
Python	3.12+
フレームワーク	FastAPI 0.116+
タスクキュー	Celery 5.5+
バリデーション	Pydantic 2.x、pydantic-settings
ロギング	FastAPIStructLogger を使用した structlog

共通スキルへのリンク

このスキルは、共通の Python スキルを拡張したものです。以下を参照してください。

• Python のパターン - データクラス、内包表記、インポート
• 非同期処理のパターン - async/await、asyncio、並行処理
• 型付けのパターン - 型ヒント、ジェネリクス、Pydantic
• テストのパターン - Pytest、フィクスチャ、モック
• エラー処理 - 例外、ロギング、コンテキストマネージャー
• セキュリティのパターン - パストラバーサル、インジェクション、シークレット

サーバー固有のチェックリスト

トピック	ファイル	主なポイント
FastAPI	fastapi.md	ルート、依存関係、ミドルウェア、例外ハンドラー
Celery	celery.md	タスクのパターン、エラー処理、リトライ、シグネチャ
Pydantic	pydantic.md	Pydantic v2 のモデル、バリデーション、シリアライズ
パフォーマンス	performance.md	非同期処理のパターン、キャッシュ、コネクションプーリング

アーキテクチャの概要

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

クイックリファレンス

設定パターン（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

構造化ロギング

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

カスタム例外

# 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)

サービスレイヤーパターン

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

レビュー用チェックリストの概要

1. FastAPI のルート（優先度：高）

   • 適切な async/sync 関数の選択
   • response_model= でレスポンスモデルを定義
   • OpenAPI メタデータ（operation_id、tags、summary）
   • 適切なステータスコードを指定した HTTPException

2. Celery タスク（優先度：高）

   • サービス間呼び出しにはシグネチャを使用
   • 適切なタイムアウトとポーリングを実装
   • タスクの失敗を適切に処理
   • 並列タスク用のグループメタデータを保存

3. Pydantic モデル（優先度：中）

   • Pydantic v2 のパターン（model_config、Field）を使用
   • フィールド制約による適切なバリデーション
   • model_dump() によるシリアライズ

4. パフォーマンス（優先度：中）

   • 非同期関数内でブロッキング呼び出しを避ける
   • 外部サービスには適切なコネクションプーリングを使用
   • 適切な箇所にキャッシュを実装

grep パターンを含む詳細なチェックリストについては、各トピックのファイルを参照してください。