6. Building a Provider

Write a platform service that other components consume.

What Is a Provider

A provider is just a component that @provides a contract other components @requires. The kernel ships several built-in providers — ConfigProvider (IConfig), LoggingProvider (ILogger), CredentialProvider (ICredentials), StorageProvider (IStorage), AuthProvider (IAuth), WorkspaceProvider (IWorkspace), TracingProvider (ITracer), APIGateway (IGateway), and the transport adapters — and you can write more. See Contracts reference for the full list.

Step 1: Define the Contract

from typing import Protocol, runtime_checkable, Any

@runtime_checkable
class ICache(Protocol):
    """Simple key-value cache."""
    def get(self, key: str) -> Any | None: ...
    def set(self, key: str, value: Any, ttl_s: int = 0) -> None: ...
    def delete(self, key: str) -> None: ...

Step 2: Implement It

import time
from signalpy.kernel import component, provides, requires, lifecycle
from signalpy.kernel.contracts import IConfig

@component("cache", version="1.0")
@provides(ICache)
@requires(config=IConfig)
class InMemoryCacheProvider:
    @lifecycle.activate
    def activate(self):
        self._store = {}
        self._default_ttl = self.rt.config.get("cache.default_ttl", 300)

    def get(self, key):
        entry = self._store.get(key)
        if entry is None: return None
        value, expires = entry
        if expires and time.time() > expires:
            del self._store[key]
            return None
        return value

    def set(self, key, value, ttl_s=0):
        ttl = ttl_s or self._default_ttl
        expires = time.time() + ttl if ttl else 0
        self._store[key] = (value, expires)

    def delete(self, key):
        self._store.pop(key, None)

    @lifecycle.deactivate
    def deactivate(self):
        self._store.clear()

Step 3: Use It

from signalpy.kernel import component, requires, runnable
from pydantic import BaseModel

class SearchParams(BaseModel):
    query: str

@component("search")
@requires(cache=ICache)
class SearchApp:
    @runnable("search", params=SearchParams, description="Search")
    async def search(self, params):
        cached = self.rt.cache.get(f"search:{params.query}")
        if cached:
            return cached
        results = await self._do_search(params.query)
        self.rt.cache.set(f"search:{params.query}", results, ttl_s=300)
        return results

Step 4: Swap It

In production, swap for Redis. Same contract, different provider:

@component("cache", version="2.0")
@provides(ICache)
@requires(config=IConfig)
class RedisCacheProvider:
    @lifecycle.activate
    def activate(self):
        import redis
        url = self.rt.config.get("cache.redis_url", "redis://localhost:6379")
        self._client = redis.Redis.from_url(url)

    def get(self, key): return self._client.get(key)
    def set(self, key, value, ttl_s=0): self._client.set(key, value, ex=ttl_s or None)
    def delete(self, key): self._client.delete(key)

    @lifecycle.deactivate
    def deactivate(self):
        self._client.close()

Zero changes to SearchApp

Swap InMemoryCacheProvider for RedisCacheProvider in kernel.discover(). Everything else stays the same.

The Bridge Pattern

The Redis provider follows a pattern:

Lifecycle: activate() → connect, deactivate() → close
Surface: get/set/delete — thin wrapper, covers 90% of use cases
Escape hatch: self._client — raw Redis object for pub/sub, Lua scripts, etc.

See Bridging External Systems for full details.