Skip to content

Plugin System

ContextRouter uses a declarative plugin architecture to extend its capabilities without modifying core code. Plugins are self-contained directories with a plugin.yaml manifest and a Python entry point. The Router loads them at startup, mediating access through a capability-gated PluginContext.

When to Use Plugins vs Manifests

ContextUnity has two complementary extension mechanisms with zero overlap:

NeedMechanism
New graph pipeline→ Project manifest YAML (contextunity.project.yaml)
New LLM provider→ Plugin with providers capability
New external tool→ Plugin with tools capability
New data source connector→ Plugin with connectors capability
New processing transformer→ Plugin with transformers capability
Custom graph topology→ inline graph (project-authored nodes/edges) or template overrides: in project manifest
Domain-specific graph (commerce, analytics)→ Manifest YAML owned by the project, never a plugin

Rule of thumb: Manifests define what graphs to run. Plugins define what capabilities the Router has.

Plugin Manifest (plugin.yaml)

Every plugin must include a plugin.yaml (or plugin.yml) manifest. The manifest is validated with Pydantic at load time — invalid manifests are rejected before any code runs.

name: "my-custom-provider"
version: "1.0.0"
description: "Adds OpenRouter LLM provider to ContextRouter"
author: "Acme Corp"
# Version compatibility
requires:
contextunity.router: ">=0.9.0"
# Declare which registries this plugin needs access to
capabilities:
- providers
# Python file with on_load(ctx) hook
entry_point: "plugin.py"
# Set to false to disable without removing
enabled: true

Manifest Fields

FieldRequiredDescription
name✅Lowercase identifier (^[a-z0-9][a-z0-9._-]*$), max 128 chars
version✅Semantic version (x.y.z)
description—Human-readable description
author—Plugin author
requires—Version constraints (e.g., contextunity.router: ">=0.9.0")
capabilities—List of capability grants (see below)
entry_point—Python file name (default: plugin.py, no path separators)
enabled—Boolean, default true

Capabilities

Capabilities control what a plugin can register. A plugin that declares capabilities: [tools] can call ctx.register_tool() but cannot call ctx.register_graph() — the PluginContext enforces this at runtime with a PermissionError.

CapabilityGrants Access ToUse Case
toolsctx.register_tool(instance)Custom LLM function-calling tools
graphsctx.register_graph(name, builder)Project graph state machines
providersctx.register_provider(name, cls)New LLM provider backends
connectorsctx.register_connector(name, cls)Data source adapters
transformersctx.register_transformer(name, cls)Processing stages (NER, classification, etc.)

A plugin can request multiple capabilities:

capabilities:
- tools
- connectors

Writing a Plugin

Directory Structure

my-plugin/
plugin.yaml # Manifest (required)
plugin.py # Entry point with on_load(ctx)

Entry Point (plugin.py)

The entry point must expose an on_load(ctx) function. ContextRouter calls it during bootstrap, passing a PluginContext that mediates all registration.

"""Custom weather tool plugin for ContextRouter."""
from contextunity.router.core.plugins import PluginContext
def on_load(ctx: PluginContext) -> None:
"""Called by Router at startup. Register capabilities here."""
from langchain_core.tools import tool
@tool
def get_weather(city: str) -> str:
"""Get current weather for a city."""
# Your implementation here
return f"Weather in {city}: 22°C, sunny"
ctx.register_tool(get_weather)

Example: Custom LLM Provider

plugin.yaml
name: "custom-llm-provider"
version: "1.0.0"
description: "Adds a custom LLM provider"
capabilities:
- providers
plugin.py
from contextunity.router.core.plugins import PluginContext
def on_load(ctx: PluginContext) -> None:
"""Register a custom LLM provider."""
class MyProvider:
"""Custom provider that wraps a local model."""
def create_llm(self, model_name: str, **kwargs):
# Return a LangChain-compatible LLM instance
...
ctx.register_provider("my-provider", MyProvider)

Example: Custom Data Connector

plugin.yaml
name: "elasticsearch-connector"
version: "1.0.0"
capabilities:
- connectors
plugin.py
from contextunity.router.core.plugins import PluginContext
def on_load(ctx: PluginContext) -> None:
"""Register an Elasticsearch data connector."""
class ElasticsearchConnector:
def fetch(self, query: str, **kwargs) -> list[dict]:
# Query Elasticsearch and return results
...
ctx.register_connector("elasticsearch", ElasticsearchConnector)

Loading and Discovery

Automatic Scanning

ContextRouter scans plugin directories listed in settings.toml:

[plugins]
paths = [
"~/my-contextrouter-plugins",
"./plugins"
]

During bootstrap, the Router:

  1. Iterates subdirectories in each configured path.
  2. Looks for plugin.yaml (or plugin.yml) in each subdirectory.
  3. Validates the manifest schema via Pydantic (fail-closed on invalid).
  4. Checks version compatibility against requires.
  5. Creates a PluginContext scoped to the manifest’s declared capabilities.
  6. Imports entry_point and calls on_load(ctx).

Plugins without a manifest are skipped — there is no legacy bare .py fallback.

Programmatic Loading

You can also load plugins explicitly:

from pathlib import Path
from contextunity.router.core.plugins import load_plugin
ctx = load_plugin(Path("./my-plugin"))
if ctx:
print(ctx.summary())
# {'name': 'my-plugin', 'version': '1.0.0',
# 'capabilities': ['tools'], 'tools': ['get_weather'],
# 'graphs': [], 'connectors': [], 'providers': [], 'transformers': []}

Security Model

Capability Isolation

Plugins operate within strict boundaries:

  • No ambient access — plugins cannot import internal Router registries directly. All registration goes through PluginContext methods.
  • Capability enforcement — attempting to register a tool without declaring tools in capabilities raises PermissionError.
  • No path traversal — entry_point must be a plain .py filename (no ../, no /).
  • Duplicate prevention — loading the same plugin name twice returns the existing instance without re-executing on_load.

What Plugins Cannot Do

  • Access environment secrets directly (use Shield for secret management)
  • Modify existing graphs registered by other plugins or manifests
  • Register capabilities not declared in their manifest
  • Execute arbitrary code in the Router process outside their on_load scope

Introspection

List all loaded plugins and what they registered:

from contextunity.router.core.plugins import get_loaded_plugins
for name, ctx in get_loaded_plugins().items():
info = ctx.summary()
print(f"{info['name']} v{info['version']}")
print(f" Capabilities: {info['capabilities']}")
print(f" Tools: {info['tools']}")
print(f" Providers: {info['providers']}")
print(f" Connectors: {info['connectors']}")

Testing Plugins

Use reset_plugins() to clear the registry between tests:

import pytest
from contextunity.router.core.plugins import (
PluginContext,
PluginManifest,
PluginCapability,
reset_plugins,
)
@pytest.fixture(autouse=True)
def _clean():
reset_plugins()
yield
reset_plugins()
def test_my_plugin():
manifest = PluginManifest(
name="test-plugin",
version="1.0.0",
capabilities=[PluginCapability.TOOLS],
)
ctx = PluginContext(manifest, plugin_dir=Path("/tmp/test"))
# Plugin code can register tools
ctx.register_tool(my_tool_instance)
# But NOT graphs (capability not granted)
with pytest.raises(PermissionError):
ctx.register_graph("test", lambda: None)