Plugin Developer Guide — Phase 7: Plug & Play

XYBEROS supports a plug & play architecture where third-party packages can register processors, providers, and strategies without modifying the core codebase. This guide explains how to create and distribute plugins.

Quick Start

Creating a custom perception processor in three steps:

Step 1 — Create your processor

# my_plugin/processors.py
from xyberos.common.pipeline import Stage
from xyberos.brain.perception.models import Observation


class UppercaseProcessor(Stage[Observation]):
    """Converts all text to uppercase — a minimal example."""

    @property
    def name(self) -> str:
        return "uppercase"

    def process(self, observation: Observation) -> Observation:
        if observation.normalized and isinstance(observation.normalized, str):
            observation.normalized = observation.normalized.upper()
        return observation

Step 2 — Register via entry points

In your package's pyproject.toml:

[project.entry-points."xyberos.processors"]
uppercase = "my_plugin.processors:UppercaseProcessor"

Step 3 — Done

When XYBEROS loads, CognitivePerceiver auto-discovers your processor via importlib.metadata entry points and includes it in the perception pipeline.

Entry-Point Discovery

XYBEROS discovers plugins through Python's standard importlib.metadata.entry_points mechanism.

Supported entry-point groups

Group	Purpose	Registered type
`xyberos.processors`	Perception pipeline stages	`Stage[Observation]`
`xyberos.providers`	Provider implementations	`Provider` subclass
`xyberos.strategies`	Reasoning strategies	Strategy Protocol
`xyberos.memory`	Memory tier providers	`IMemoryStore`

Manual discovery

You can also discover plugins programmatically:

from xyberos.common.registry import discover_entry_points

processors = discover_entry_points("xyberos.processors")
for name, cls in processors.items():
    print(f"Found processor: {name} -> {cls}")

Or register them into an existing registry:

from xyberos.common.registry import ProviderRegistry, discover_and_register

registry = ProviderRegistry()
discover_and_register(registry, "xyberos.providers", default_name="my_provider")

Discovery order

Decorator-based — @perception_processor decorated classes are always checked first.
Entry-point-based — pyproject.toml entry points are loaded second (when use_entry_points=True, the default).

Entry points do not override decorator-registered processors with the same name.

Capability-Based Selection

Every provider can declare its capability and implement can_handle() to enable intelligent selection.

Declaring a capability

from xyberos.common import Provider, ProviderCapability, ml_capability
from xyberos.common.capabilities import deterministic_capability


class MyMLProvider(Provider):
    @property
    def name(self) -> str:
        return "my_ml"

    @property
    def capability(self) -> ProviderCapability:
        return ml_capability(
            languages=("en",),
            latency_p95_ms=150.0,
            description="ML-based intent classifier",
        )

    def can_handle(self, input: str) -> bool:
        """Only handle English inputs over 10 characters."""
        return len(input) > 10 and isinstance(input, str)

Capability phases

Phase	Label	Typical latency	Examples
1	Deterministic	< 1ms	Regex, language detection, keyword counting
2	NLP	10-100ms	spaCy POS tagging, dependency parsing
3	ML	50-500ms	Classifiers, sentence transformers
4	LLM	500ms+	LLM-gated reasoning, fallback

Selecting the best provider

from xyberos.common.registry import select_best_provider

# With max_phase=2, only deterministic and NLP providers are considered
best = select_best_provider(providers, input_text, max_phase=2)

# prefer_fast=True picks the lowest-latency provider
fastest = select_best_provider(providers, input_text, prefer_fast=True)

Or on a registry directly:

best = registry.best(input_text, max_phase=2)

Graceful Degradation

The graceful degradation pattern ensures that if the preferred provider fails, the system falls back to the next capable provider.

Using `fallback_execute`

from xyberos.common.registry import fallback_execute

result = fallback_execute(
    providers,          # list of provider instances
    input_value,        # input to check and process
    "detect",           # method name to call
    max_phase=2,        # optional: limit to deterministic/NLP
    default_result=("unknown", 0.0),  # fallback if all fail
)

Using `registry.fallback()`

result = registry.fallback(
    input_value,
    "detect",
    default_result=("unknown", 0.0),
)

GracefulPerceiver

The GracefulPerceiver extends CognitivePerceiver with automatic degradation:

from xyberos.brain.perception.perceiver import GracefulPerceiver
from xyberos.brain.perception.models import PerceptionInput

perceiver = GracefulPerceiver(fallback_phase=2)  # fallback to NLP

# If full pipeline fails, automatically retries with phase ≤ 2
# If that also fails, returns a minimal Observation with error metadata
observation = perceiver.perceive(PerceptionInput(content="Hello"))

Manager-level fallback

Managers like IntentManager and SafetyManager now use fallback:

manager = IntentManager(registry)

# Tries default first, falls through on failure
intent, confidence = manager.detect("hello world")
# Returns ("unknown", 0.0) if all providers fail

# Fast path — no fallback, direct default provider
intent, confidence = manager.detect_fast("hello world")

Reference

`ProviderCapability` fields

Field	Type	Default	Description
`phase`	`int`	`1`	1=deterministic, 2=NLP, 3=ML, 4=LLM
`languages`	`tuple[str, ...]`	`()`	Supported ISO language codes (empty = any)
`min_input_length`	`int`	`0`	Minimum characters required
`max_input_length`	`int`	`0`	Maximum characters supported (0 = unlimited)
`latency_p95_ms`	`float`	`0.0`	Expected p95 latency
`requires_gpu`	`bool`	`False`	Requires GPU hardware
`requires_network`	`bool`	`False`	Requires network access
`description`	`str`	`""`	Human-readable description
`metadata`	`dict`	`{}`	Arbitrary extension data

Factory helpers

Function	Phase	Default latency
`deterministic_capability()`	1	1ms
`nlp_capability()`	2	50ms
`ml_capability()`	3	200ms
`llm_capability()`	4	2000ms

API surface

Function / Method	Description
`discover_entry_points(group)`	Discover entry-point registered classes
`discover_and_register(registry, group)`	Discover and register into a registry
`select_best_provider(providers, input)`	Select best provider by capability
`fallback_execute(providers, input, method)`	Execute with graceful degradation
`registry.discover(group)`	Discover and register plugins
`registry.best(input)`	Select best registered provider
`registry.fallback(input, method)`	Execute with fallback across registry
`service.best(input)`	Select best provider via service
`service.fallback(input, method)`	Execute with fallback via service
`provider.can_handle(input)`	Check if provider can process input
`provider.capability`	Get provider's capability declaration

Migration Guide

See docs/migration-guide-phase7.md for instructions on migrating existing code to Phase 7 patterns.