rdst-services

RDST Service Creation Guide

This skill provides guidance for creating new services in rdst/lib/services/.

Why Services Exist

Services are the single source of truth for business logic shared between:

• CLI commands - via asyncio.run() bridge with Renderer for terminal output
• Web API - via native async with SSE (Server-Sent Events) streaming
• MCP server - via async interface for Claude Code integration

This architecture ensures:

• No logic duplication between CLI and web
• Consistent behavior across all interfaces
• Testable, isolated business logic
• Type-safe event contracts

Service Patterns

RDST uses two primary patterns:

1. Async Generator Services (Streaming)

Use for operations that:

• Take time and benefit from progress updates
• Need to stream results incrementally
• May require user interaction mid-execution

# my_service.py
from typing import AsyncGenerator
from .types import MyEvent, MyInput, MyOptions

class MyService:
    """Stateless service for [description]."""

    def __init__(self) -> None:
        """Initialize (no constructor parameters - stateless)."""
        pass

    async def do_thing(
        self,
        input: MyInput,
        options: MyOptions,
    ) -> AsyncGenerator[MyEvent, None]:
        """Execute operation and yield events.

        Yields:
            MyStatusEvent: Progress updates
            MyResultEvent: Final results
            MyErrorEvent: On failure
        """
        try:
            yield MyStatusEvent(type="status", message="Starting...")
            
            # Do work...
            result = await self._perform_work(input)
            
            yield MyResultEvent(type="result", data=result)
            
        except Exception as e:
            yield MyErrorEvent(type="error", message=str(e))

2. Synchronous Services (Return-based)

Use for operations that:

• Complete quickly
• Don't need streaming feedback
• Return structured results

# my_service.py
from .types import MyResult, MyInput

class MyService:
    """Stateless service for [description]."""

    def __init__(self) -> None:
        pass

    def get_status(self) -> MyResult:
        """Check status and return result."""
        # Do work...
        return MyResult(
            success=True,
            data={"key": "value"}
        )

    def perform_action(self, input: MyInput) -> MyResult:
        """Perform action and return result."""
        try:
            # Do work...
            return MyResult(success=True, message="Done")
        except Exception as e:
            return MyResult(success=False, error=str(e))