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 providerPlugin with providers capability
New external toolPlugin with tools capability
New data source connectorPlugin with connectors capability
New processing transformerPlugin 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
nameLowercase identifier (^[a-z0-9][a-z0-9._-]*$), max 128 chars
versionSemantic version (x.y.z)
descriptionHuman-readable description
authorPlugin author
requiresVersion constraints (e.g., contextunity.router: ">=0.9.0")
capabilitiesList of capability grants (see below)
entry_pointPython file name (default: plugin.py, no path separators)
enabledBoolean, 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 traversalentry_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)