Skip to content
Back to top
View this page

# Protocol-Based Typed Service Access

MindRoot provides a Protocol-based system for typed service access, enabling IDE autocomplete and type checking when calling services.

## Overview

MindRoot services are dynamically registered and called via `service_manager`. While flexible, this loses type information. The Protocols system restores type safety:

    from mindroot.protocols import llm, image, tts

    # Full IDE autocomplete!
    stream = await llm.stream_chat('gpt-4', messages=[...], context=ctx)
    result = await image.generate('a red dragon', width=1024, context=ctx)
    audio = await tts.synthesize('Hello world', context=ctx)

## Quick Start

### Using Pre-instantiated Proxies (Recommended)

The simplest way to use typed services:

    from mindroot.protocols import llm, image, tts, stt, web_search

    # These are lazy proxies - they initialize on first use
    stream = await llm.stream_chat('gpt-4', messages=[...], context=ctx)
    result = await image.generate('a red dragon', context=ctx)

### Creating Your Own Typed Proxy

Alternatively, create a typed proxy explicitly:

    from mindroot.protocols import LLM
    from mindroot.services import service_manager

    llm: LLM = service_manager.typed(LLM)
    stream = await llm.stream_chat('gpt-4', messages=[...], context=ctx)

## Implementing Services with Protocols

### Class-Based Implementation (Recommended)

Use `@service_class` to implement a Protocol with full IDE autocomplete:

    from mindroot.services import service_class
    from mindroot.protocols import LLM
    from typing import AsyncIterator, Any

    @service_class(LLM)
    class MyLLM(LLM):
        """My LLM implementation."""

        async def stream_chat(
            self,
            model: str,
            messages: list = None,
            context: Any = None,
            num_ctx: int = 200000,
            temperature: float = 0.0,
            max_tokens: int = 5000,
            num_gpu_layers: int = 0
        ) -> AsyncIterator[str]:
            # IDE autocomplete works here!
            ...

        async def chat(self, model: str, messages: list,
                       context: Any = None) -> str:
            ...

        async def format_image_message(self, pil_image: Any,
                                       context: Any = None) -> dict:
            ...

        async def get_service_models(self, context: Any = None) -> dict:
            return {
                "stream_chat": ["my-model-1", "my-model-2"],
                "chat": ["my-model-1", "my-model-2"]
            }

The `@service_class` decorator will:

1. Instantiate the class (must have no required `__init__` args)

2. Find all methods defined in the Protocol

3. Register each matching method as a service

### Function-Based Implementation (Legacy)

The traditional `@service()` decorator still works for standalone functions:

    from mindroot.services import service

    @service()
    async def stream_chat(model, messages=[], context=None, ...):
        ...

This approach is fully backwards compatible but doesn’t provide IDE autocomplete for the function signature.

## Available Protocols

Core MindRoot provides these common Protocol definitions:

### LLM (Language Model)

    class LLM(Protocol):
        async def stream_chat(self, model: str, messages: list = None,
                              context = None, temperature: float = 0.0,
                              max_tokens: int = 5000) -> AsyncIterator[str]: ...

        async def chat(self, model: str, messages: list,
                       context = None) -> str: ...

        async def format_image_message(self, pil_image, context = None) -> dict: ...

        async def get_service_models(self, context = None) -> dict: ...

Implemented by: `ah_openai`, `ah_anthropic`, `ah_ollama`, `mr_deepseek`, etc.

### Image

    class Image(Protocol):
        async def generate(self, prompt: str, width: int = 1024,
                          height: int = 1024, context = None) -> dict: ...

Implemented by: `ah_flux`, `ah_sd`, `mr_imagen`, etc.

### TTS (Text-to-Speech)

    class TTS(Protocol):
        async def synthesize(self, text: str, voice: str = 'default',
                            context = None) -> bytes: ...

        async def list_voices(self, context = None) -> list: ...

Implemented by: `ah_tts`, `mr_eleven_stream`, `mr_f5_tts`, etc.

### STT (Speech-to-Text)

    class STT(Protocol):
        async def transcribe(self, audio: bytes, context = None) -> str: ...

        async def transcribe_stream(self, audio_stream: AsyncIterator[bytes],
                                   context = None) -> AsyncIterator[str]: ...

Implemented by: `mr_deepgram`, whisper plugins, etc.

### WebSearch

    class WebSearch(Protocol):
        async def search(self, query: str, num_results: int = 10,
                        context = None) -> list: ...

## Plugin-Defined Protocols

Plugins can define their own Protocols without modifying core MindRoot:

### Defining a Protocol

    # In mr_sip/protocols.py
    from typing import Protocol, runtime_checkable

    @runtime_checkable
    class SIP(Protocol):
        """SIP telephony service protocol."""

        async def dial_service(self, destination: str,
                              context = None) -> dict: ...

        async def end_call_service(self, context = None) -> dict: ...

        async def sip_audio_out_chunk(self, audio_chunk: bytes,
                                      context = None) -> None: ...

### Implementing with service_class

    # In mr_sip/mod.py
    from mindroot.services import service_class
    from .protocols import SIP

    @service_class(SIP)
    class MySIPProvider(SIP):
        async def dial_service(self, destination: str, context = None) -> dict:
            # Implementation with IDE autocomplete
            ...

### Creating a Pre-instantiated Proxy

    # In mr_sip/__init__.py or mr_sip/protocols.py
    from mindroot.lib.providers.protocols.registry import create_lazy_proxy
    from .protocols import SIP

    # Create a lazy proxy for convenient access
    sip: SIP = create_lazy_proxy(SIP)

### Using Plugin Protocols

    # Users can then import and use:
    from mr_sip.protocols import sip

    result = await sip.dial_service('555-1234', context=ctx)
    await sip.end_call_service(context=ctx)

## Protocol Registry

Plugins can optionally register their Protocols for discovery:

### Registering a Protocol

    from mindroot.protocols import register_protocol
    from .protocols import SIP

    register_protocol('sip', SIP)

### Discovering Protocols

    from mindroot.protocols import get_protocol, list_protocols

    # List all registered protocols
    protocols = list_protocols()  # {'llm': LLM, 'image': Image, ...}

    # Get a specific protocol
    SIP = get_protocol('sip')
    if SIP:
        from mindroot.services import service_manager
        sip = service_manager.typed(SIP)

### Method-to-Service Mapping

When a Protocol method name differs from the service name, use explicit mapping:

    from mindroot.protocols import map_method_to_service, Image

    # Map Image.generate() to the 'image' service
    map_method_to_service(Image, 'generate', 'image')

## Backwards Compatibility

The Protocol system is fully backwards compatible. All existing code continues to work:

    # This still works exactly as before
    from mindroot.services import service_manager

    stream = await service_manager.stream_chat('gpt-4', messages=[...], context=ctx)
    result = await service_manager.image('a red dragon', context=ctx)

    # Legacy imports also work
    from lib.providers.services import service_manager
    from lib.providers.protocols import llm

## Import Paths

MindRoot supports both short and full import paths:

    # Short paths (recommended)
    from mindroot.services import service, service_class, service_manager
    from mindroot.protocols import LLM, Image, TTS, llm, image, tts

    # Full paths (also work)
    from mindroot.lib.providers.services import service, service_class, service_manager
    from mindroot.lib.providers.protocols import LLM, Image, TTS, llm, image, tts

## API Reference

### @service_class(protocol)

Class decorator that registers all Protocol methods as services.

param protocol:

  The Protocol class being implemented

param flags:

  Optional flags to pass to each service registration

    from mindroot.services import service_class
    from mindroot.protocols import LLM

    @service_class(LLM)
    class MyLLM(LLM):
        async def stream_chat(self, model: str, ...) -> AsyncIterator[str]:
            ...

### @service()

Function decorator for registering individual service functions.

param flags:

  Optional flags for the service

    from mindroot.services import service

    @service()
    async def my_function(arg1, context=None):
        ...

### service_manager.typed(protocol)

Get a typed proxy for a service protocol.

param protocol:

  A Protocol class defining the service interface

returns:

  A proxy object typed as the Protocol

    from mindroot.protocols import LLM
    from mindroot.services import service_manager

    llm: LLM = service_manager.typed(LLM)

### service_manager.get_protocol(name)

Get a registered Protocol by name.

param name:

  The protocol name (e.g., ‘llm’, ‘sip’)

returns:

  The Protocol class, or None if not found

    SIP = service_manager.get_protocol('sip')

### service_manager.list_protocols()

List all registered Protocols.

returns:

  Dict mapping protocol names to Protocol classes

    protocols = service_manager.list_protocols()
    # {'llm': LLM, 'image': Image, 'tts': TTS, ...}

### create_lazy_proxy(protocol)

Create a lazy typed proxy for a Protocol. Useful for plugins.

param protocol:

  The Protocol class

returns:

  A LazyTypedProxy typed as the Protocol

    from mindroot.lib.providers.protocols.registry import create_lazy_proxy

    sip: SIP = create_lazy_proxy(SIP)

### register_protocol(name, protocol_class)

Register a Protocol class for discovery.

param name:

  A unique name for the protocol

param protocol_class:

  The Protocol class

    from mindroot.protocols import register_protocol

    register_protocol('sip', SIP)

### map_method_to_service(protocol, method_name, service_name)

Create an explicit mapping from a Protocol method to a service name.

param protocol:

  The Protocol class

param method_name:

  The method name in the Protocol

param service_name:

  The actual service name to call

    from mindroot.protocols import map_method_to_service, Image

    map_method_to_service(Image, 'generate', 'image')