Service Registry

Table of Contents

• Overview
• When to Use
• Core Concepts
• Service Configuration
• Execution Result
• Quick Start
• Register Services
• Execute via Service
• Health Checks
• Service Selection
• Auto-Selection
• Failover Pattern
• Integration Pattern
• Detailed Resources
• Exit Criteria

Service Registry

Overview

A registry pattern for managing connections to external services. Handles configuration, health checking, and execution across multiple service integrations.

When To Use

• Managing multiple external services.
• Need consistent execution interface.
• Want health monitoring across services.
• Building service failover logic.

When NOT To Use

• Single service integration without registry needs

Core Concepts

Service Configuration

@dataclass
class ServiceConfig:
    name: str
    command: str
    auth_method: str  # "api_key", "oauth", "token"
    auth_env_var: str
    quota_limits: dict
    models: list[str] = field(default_factory=list)

Verification: Run the command with --help flag to verify availability.

Execution Result

@dataclass
class ExecutionResult:
    success: bool
    stdout: str
    stderr: str
    exit_code: int
    duration: float
    tokens_used: int

Verification: Run the command with --help flag to verify availability.

Quick Start

Register Services

from leyline.service_registry import ServiceRegistry

registry = ServiceRegistry()

registry.register("gemini", ServiceConfig(
    name="gemini",
    command="gemini",
    auth_method="api_key",
    auth_env_var="GEMINI_API_KEY",
    quota_limits={"rpm": 60, "daily": 1000}
))

Verification: Run the command with --help flag to verify availability.

Execute via Service

result = registry.execute(
    service="gemini",
    prompt="Analyze this code",
    files=["src/main.py"],
    model="gemini-2.5-pro"
)

if result.success:
    print(result.stdout)

Verification: Run the command with --help flag to verify availability.

Health Checks

# Check single service
status = registry.health_check("gemini")

# Check all services
all_status = registry.health_check_all()
for service, healthy in all_status.items():
    print(f"{service}: {'OK' if healthy else 'FAILED'}")

Verification: Run the command with --help flag to verify availability.

Service Selection

Auto-Selection

# Select best service for task
service = registry.select_service(
    requirements={
        "large_context": True,
        "fast_response": False
    }
)

Verification: Run the command with --help flag to verify availability.

Failover Pattern

def execute_with_failover(prompt: str, files: list) -> ExecutionResult:
    for service in registry.get_healthy_services():
        result = registry.execute(service, prompt, files)
        if result.success:
            return result
    raise AllServicesFailedError()

Verification: Run the command with --help flag to verify availability.

Integration Pattern

# In your skill's frontmatter
dependencies: [leyline:service-registry]

Verification: Run the command with --help flag to verify availability.

Detailed Resources

• Service Config: See modules/service-config.md for configuration options.
• Execution Patterns: See modules/execution-patterns.md for advanced usage.

Exit Criteria

• Services registered with configuration.
• Health checks passing.
• Execution results properly handled.

Execution Patterns

Command Building

Template Expansion

def build_command(
    config: ServiceConfig,
    prompt: str,
    files: list[str],
    model: str = None
) -> str:
    """Build command from template."""
    file_args = " ".join(f"@{f}" for f in files)
    model = model or config.default_model

    return config.command_template.format(
        command=config.command,
        prompt=prompt,
        files=file_args,
        model=model
    )

Safe Execution

import subprocess
import shlex

def execute_safely(command: str, timeout: int = 60) -> ExecutionResult:
    """Execute command with safety measures."""
    start = time.time()

    try:
        result = subprocess.run(
            shlex.split(command),
            capture_output=True,
            text=True,
            timeout=timeout
        )

        return ExecutionResult(
            success=(result.returncode == 0),
            stdout=result.stdout,
            stderr=result.stderr,
            exit_code=result.returncode,
            duration=time.time() - start,
            tokens_used=estimate_tokens(result.stdout)
        )
    except subprocess.TimeoutExpired:
        return ExecutionResult(
            success=False,
            stderr="Command timed out",
            exit_code=-1,
            duration=timeout
        )

Retry Patterns

Exponential Backoff

def execute_with_retry(
    registry: ServiceRegistry,
    service: str,
    prompt: str,
    max_retries: int = 3
) -> ExecutionResult:
    """Execute with exponential backoff retry."""
    for attempt in range(max_retries):
        result = registry.execute(service, prompt)

        if result.success:
            return result

        if "rate limit" in result.stderr.lower():
            wait_time = 2 ** attempt  # 1, 2, 4 seconds
            time.sleep(wait_time)
        else:
            break  # Non-retryable error

    return result

Circuit Breaker

class CircuitBreaker:
    def __init__(self, failure_threshold: int = 5, reset_timeout: int = 60):
        self.failures = 0
        self.threshold = failure_threshold
        self.reset_timeout = reset_timeout
        self.last_failure = 0
        self.state = "closed"  # closed, open, half-open

    def can_execute(self) -> bool:
        if self.state == "closed":
            return True
        if self.state == "open":
            if time.time() - self.last_failure > self.reset_timeout:
                self.state = "half-open"
                return True
            return False
        return True  # half-open allows one attempt

    def record_result(self, success: bool):
        if success:
            self.failures = 0
            self.state = "closed"
        else:
            self.failures += 1
            self.last_failure = time.time()
            if self.failures >= self.threshold:
                self.state = "open"

Parallel Execution

Multi-Service Query

import asyncio

async def execute_parallel(
    registry: ServiceRegistry,
    services: list[str],
    prompt: str
) -> dict[str, ExecutionResult]:
    """Execute same prompt across multiple services."""
    tasks = {
        service: asyncio.create_task(
            registry.execute_async(service, prompt)
        )
        for service in services
    }

    results = {}
    for service, task in tasks.items():
        results[service] = await task

    return results

Service Configuration

Configuration Schema

Full Configuration

@dataclass
class ServiceConfig:
    # Identity
    name: str
    display_name: str = ""

    # Execution
    command: str  # CLI command or endpoint
    command_template: str = "{command} -p {prompt}"

    # Authentication
    auth_method: str  # api_key, oauth, token, none
    auth_env_var: str = ""
    auth_check_cmd: str = ""  # Command to verify auth

    # Quotas
    quota_limits: dict = field(default_factory=dict)
    # Example: {"rpm": 60, "tpm": 100000, "daily": 1000}

    # Models
    models: list[str] = field(default_factory=list)
    default_model: str = ""

    # Capabilities
    max_context: int = 100000
    supports_files: bool = True
    supports_streaming: bool = False

    # Health
    health_check_cmd: str = ""
    timeout_seconds: int = 60

Service Examples

Gemini Service

GEMINI_CONFIG = ServiceConfig(
    name="gemini",
    display_name="Google Gemini",
    command="gemini",
    command_template="gemini -p '{prompt}' {files}",
    auth_method="api_key",
    auth_env_var="GEMINI_API_KEY",
    auth_check_cmd="gemini auth status",
    quota_limits={
        "rpm": 60,
        "tpm": 1000000,
        "daily": 1000
    },
    models=["gemini-2.5-flash-exp", "gemini-2.5-pro-exp"],
    default_model="gemini-2.5-flash-exp",
    max_context=1000000,
    health_check_cmd="gemini 'ping'",
    timeout_seconds=60
)

Qwen Service

QWEN_CONFIG = ServiceConfig(
    name="qwen",
    display_name="Alibaba Qwen",
    command="qwen",
    command_template="qwen -p '{prompt}' {files}",
    auth_method="api_key",
    auth_env_var="QWEN_API_KEY",
    quota_limits={
        "rpm": 120,
        "tpm": 2000000,
        "daily": 2000
    },
    models=["qwen-turbo", "qwen-max"],
    default_model="qwen-turbo",
    max_context=100000
)

Configuration Loading

From YAML

# ~/.claude/leyline/services.yaml
services:
  gemini:
    command: gemini
    auth_method: api_key
    auth_env_var: GEMINI_API_KEY
    quota_limits:
      rpm: 60
      daily: 1000

From Environment

def load_from_env(service_name: str) -> ServiceConfig:
    prefix = service_name.upper()
    return ServiceConfig(
        name=service_name,
        command=os.getenv(f"{prefix}_COMMAND", service_name),
        auth_env_var=f"{prefix}_API_KEY",
        # ... other fields
    )