CachiBot: Skill for Claude Code

CachiBot API Route Creation

Add new REST API endpoints following CachiBot's FastAPI patterns with Pydantic models, auth, and bot-scoping.

Architecture Overview

• Framework: FastAPI with APIRouter per domain
• Auth: JWT-based via require_bot_access dependency
• Models: Pydantic BaseModel for request/response schemas
• Storage: Repository pattern with PostgreSQL (SQLAlchemy 2.0 + asyncpg)
• Registration: Routers included in server.py

Step-by-Step Process

1. Define Pydantic Models

Create or extend models in cachibot/models/<domain>.py:

"""
<Domain> Models

Pydantic schemas for <domain> API.
"""

from pydantic import BaseModel


class YourItemCreate(BaseModel):
    """Request body for creating an item."""

    name: str
    description: str = ""
    # Add fields as needed


class YourItemUpdate(BaseModel):
    """Request body for updating an item (all fields optional)."""

    name: str | None = None
    description: str | None = None


class YourItemResponse(BaseModel):
    """Response model for an item."""

    id: str
    botId: str  # camelCase for frontend
    name: str
    description: str
    createdAt: str
    updatedAt: str

    @classmethod
    def from_db(cls, row: dict) -> "YourItemResponse":
        """Convert a database row to a response model."""
        return cls(
            id=row["id"],
            botId=row["bot_id"],
            name=row["name"],
            description=row["description"] or "",
            createdAt=row["created_at"],
            updatedAt=row["updated_at"],
        )

2. Create the Route File

Create cachibot/api/routes/<domain>.py:

"""
<Domain> API Routes

Endpoints for managing <domain resources>.
"""

from fastapi import APIRouter, Depends, HTTPException
from pydantic import BaseModel

from cachibot.api.auth import require_bot_access
from cachibot.models.auth import User
from cachibot.storage.repository import YourRepository

router = APIRouter(prefix="/api/bots/{bot_id}/<domain>", tags=["<domain>"])

# Repository instance
repo = YourRepository()


@router.get("")
async def list_items(
    bot_id: str,
    user: User = Depends(require_bot_access),
) -> list[YourItemResponse]:
    """List all items for a bot."""
    items = await repo.get_items_by_bot(bot_id)
    return [YourItemResponse.from_db(item) for item in items]


@router.post("", status_code=201)
async def create_item(
    bot_id: str,
    req: YourItemCreate,
    user: User = Depends(require_bot_access),
) -> YourItemResponse:
    """Create a new item."""
    import uuid
    from datetime import datetime, timezone

    now = datetime.now(timezone.utc).isoformat()
    item_id = str(uuid.uuid4())

    item = {
        "id": item_id,
        "bot_id": bot_id,
        "name": req.name,
        "description": req.description,
        "created_at": now,
        "updated_at": now,
    }
    await repo.save_item(item)
    return YourItemResponse.from_db(item)


@router.get("/{item_id}")
async def get_item(
    bot_id: str,
    item_id: str,
    user: User = Depends(require_bot_access),
) -> YourItemResponse:
    """Get a specific item."""
    item = await repo.get_item(item_id)
    if item is None or item["bot_id"] != bot_id:
        raise HTTPException(status_code=404, detail="Item not found")
    return YourItemResponse.from_db(item)


@router.put("/{item_id}")
async def update_item(
    bot_id: str,
    item_id: str,
    req: YourItemUpdate,
    user: User = Depends(require_bot_access),
) -> YourItemResponse:
    """Update an item."""
    item = await repo.get_item(item_id)
    if item is None or item["bot_id"] != bot_id:
        raise HTTPException(status_code=404, detail="Item not found")

    updates = req.model_dump(exclude_unset=True)
    if updates:
        from datetime import datetime, timezone
        updates["updated_at"] = datetime.now(timezone.utc).isoformat()
        await repo.update_item(item_id, updates)

    updated = await repo.get_item(item_id)
    return YourItemResponse.from_db(updated)


@router.delete("/{item_id}", status_code=204)
async def delete_item(
    bot_id: str,
    item_id: str,
    user: User = Depends(require_bot_access),
) -> None:
    """Delete an item."""
    item = await repo.get_item(item_id)
    if item is None or item["bot_id"] != bot_id:
        raise HTTPException(status_code=404, detail="Item not found")
    await repo.delete_item(item_id)