Contributing Guide

from fastapi import APIRouter, Depends, HTTPException
from sqlalchemy.ext.asyncio import AsyncSession
from ..database import get_db
from ..users import current_active_user
from ..models_auth import User

router = APIRouter(prefix="/api/your-feature", tags=["your-feature"])

@router.get("/")
async def list_items(
    current_user: User = Depends(current_active_user),
    db: AsyncSession = Depends(get_db)
):
    """List all items for the current user."""
    # Implementation here
    pass

from pydantic import BaseModel, ConfigDict
from uuid import UUID
from datetime import datetime

class YourFeatureCreate(BaseModel):
    name: str

class YourFeatureResponse(BaseModel):
    model_config = ConfigDict(from_attributes=True)
    id: UUID
    name: str
    created_at: datetime

from .routers import your_feature
app.include_router(your_feature.router)

import pytest
from httpx import AsyncClient

@pytest.mark.asyncio
async def test_list_items(client: AsyncClient, auth_headers: dict):
    response = await client.get("/api/your-feature", headers=auth_headers)
    assert response.status_code == 200

orchestrator/app/agent/tools/
+-- your_tool/
    +-- __init__.py
    +-- implementation.py

from ..registry import Tool, ToolCategory
from ..output_formatter import success_output, error_output

async def your_tool_executor(params, context):
    file_path = params.get("file_path")
    if not file_path:
        return error_output(
            message="file_path is required",
            suggestion="Provide the path to the file"
        )

    # Use orchestrator for file/shell operations
    from ....services.orchestration import get_orchestrator
    orchestrator = get_orchestrator()

    try:
        result = await orchestrator.read_file(
            user_id=context["user_id"],
            project_id=str(context["project_id"]),
            container_name=context.get("container_name"),
            file_path=file_path,
            project_slug=context.get("project_slug"),
            subdir=context.get("container_directory")
        )
        return success_output(message="Success", result=result)
    except Exception as e:
        return error_output(message=f"Failed: {str(e)}")

def register_your_tools(registry):
    registry.register(Tool(
        name="your_tool",
        description="What this tool does (shown to the LLM)",
        parameters={
            "type": "object",
            "properties": {
                "file_path": {
                    "type": "string",
                    "description": "Path relative to project root"
                }
            },
            "required": ["file_path"]
        },
        executor=your_tool_executor,
        category=ToolCategory.FILE_OPS,
        examples=['{"tool_name": "your_tool", "parameters": {"file_path": "src/App.tsx"}}']
    ))

@pytest.mark.asyncio
async def test_your_tool_success():
    params = {"file_path": "src/App.jsx"}
    context = {"user_id": "test", "project_id": "test", "project_slug": "test"}

    with patch('...get_orchestrator') as mock:
        mock.return_value.read_file = AsyncMock(return_value="content")
        result = await your_tool_executor(params, context)
        assert result["success"] is True

cd orchestrator
alembic revision --autogenerate -m "description_of_changes"

alembic upgrade head

alembic downgrade -1   # Rollback
alembic upgrade head   # Re-apply

git checkout -b feature/your-feature-name