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:

Instantiate the class (must have no required __init__ args)
Find all methods defined in the Protocol
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)¶

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')