Plugin API Reference

This section documents the built-in plugins provided by Genro Routes.

For additional reference, see:

Plugin Development Guide - Create custom plugins
Plugin Configuration Guide - Runtime configuration

External plugin API

from genro_routes.plugins import BasePlugin, MethodEntry

These are the supported extension types: subclass BasePlugin and use MethodEntry for handler metadata in plugin hooks. See the custom plugin example for registration and attachment. The legacy _base_plugin imports refer to the same classes and remain available for compatibility.

PydanticPlugin: strict validation and `_coerce`

Input validation is strict: an argument must already have the annotated type, so "12" for an int parameter raises ValidationError instead of being converted. An int is still accepted for a float parameter, which is Pydantic’s own strict-mode allowance.

Conversion is requested per call with the reserved keyword _coerce=True on the node call. It applies to every parameter of that call and is always consumed by the router, never reaching the handler.

svc.route.node("count")(12)                  # OK
svc.route.node("count")("12")                # ValidationError
svc.route.node("count")("12", _coerce=True)  # OK -> 12

Situation	Result
`_coerce=True`, router without the `pydantic` plugin	raises the exception mapped to `not_available` (`NotAvailable` by default)
`_coerce=True`, entry with no type hints or `pydantic_disabled=True`	silently ignored: nothing to convert
`_coerce=False` or absent	strict validation, the default

Annotated[int, Field(strict=False)] opts a single parameter out of strict mode for every call, with no _coerce.

Auto-Generated Plugin API

AuthPlugin - Authorization plugin with tag-based access control.

This plugin provides tag-based authorization for router entries. It evaluates authorization rules defined on endpoints against user tags.

Usage:

from genro_routes import RoutingClass, route

class MyAPI(RoutingClass):
    def __init__(self):
        self.route.plug("auth")

    @route(auth_rule="admin&internal")
    def admin_action(self):
        return "admin only"

    @route(auth_rule="public")
    def public_action(self):
        return "public"

# Query with user's tags (comma-separated list of tags the user has)
obj = MyAPI()
obj.route.node("admin_action", auth_tags="admin,internal")  # user has both tags
obj.route.nodes(auth_tags="admin")           # user has admin tag
obj.route.nodes(auth_tags="admin,public")    # user has admin AND public tags

Rule syntax (on entry auth_rule):

| : OR (user must have at least one)
& : AND (user must have all)
! : NOT (user must not have)
() : grouping

NOTE: Comma is NOT allowed in auth_rule. Use | for OR, & for AND.

Comma in auth_tags means the user has multiple tags (always AND).

Example: auth_rule=”admin|manager” means “user must have admin OR manager” Example: auth_rule=”admin&!guest” means “user must have admin AND NOT guest”

class genro_routes.plugins.auth.AuthPlugin(router, **config)[source]

Bases: BasePlugin

Authorization plugin with tag-based access control.

Evaluates boolean rule expressions against user tags to control access to router entries. Rules are defined per-entry via @route(auth_rule=...) and checked at runtime against tags provided via auth_tags parameter.

Rule syntax (on entry):

| : OR (user must have at least one tag)
& : AND (user must have all tags)
! : NOT (user must not have tag)
() : grouping for complex expressions

User tags (on query):

Comma-separated string of tags the user possesses. The comma means the user has ALL those tags (implicit AND).

plugin_code: “auth” - used for registration and config prefix.

plugin_description: Human-readable description.

Example

Entry definition:

@route(auth_rule="admin|manager")  # OR
def sensitive_action(self): ...

@route(auth_rule="admin&!guest")   # AND + NOT
def admin_only(self): ...

Query with user tags:

# User has admin tag -> can access admin|manager entries
router.node("sensitive_action", auth_tags="admin")

# User has both admin and hr tags
router.nodes(auth_tags="admin,hr")

Parameters:

router (Any)
config (Any)

plugin_code: str = 'auth'

plugin_description: str = 'Authorization plugin with tag-based access control'

plugin_default_param: str | None = 'rule'

configure(*, rule='', enabled=True, _target='_all_', flags=None)[source]

Define authorization rule for this entry/router.

Parameters:

rule (str) – Boolean rule expression (e.g., “admin&internal”, “!guest”). Use | for OR, & for AND. Comma is not allowed.
enabled (bool) – Whether the plugin is enabled (default True)
_target (str) – Internal - target bucket name
flags (str | None) – Internal - flag string

Raises:

ValueError – If rule contains comma (use | for OR instead).

Return type:

None

deny_reason(entry, **filters)[source]

Filter entries based on authorization rule.

Parameters:

entry (MethodEntry | RouterInterface) – MethodEntry or Router being checked.
**filters (Any) – May contain tags with user’s tags.

Returns:

Access allowed (no reason to deny). “not_authenticated”: Entry requires tags but none provided. “not_authorized”: Tags provided but don’t match rule.

Return type:

””

name

EnvPlugin - Environment capability-based access control plugin.

This plugin provides capability-based filtering for router entries. It evaluates capability requirements defined on endpoints against system capabilities.

Capabilities can come from two sources:

Request capabilities: Passed explicitly via env_capabilities parameter
Instance capabilities: Declared on RoutingClass instances via a CapabilitiesSet subclass

When traversing a router hierarchy, capabilities are accumulated from all RoutingClass instances along the path. This allows child services to inherit capabilities from their parents while adding their own.

Usage:

from genro_routes import Router, RoutingClass, route
from genro_routes.plugins.env import CapabilitiesSet, capability

class MyCapabilities(CapabilitiesSet):
    @capability
    def redis(self) -> bool:
        return True  # Check if redis is available

    @capability
    def pyjwt(self) -> bool:
        return "pyjwt" in sys.modules

class MyAPI(RoutingClass):
    def __init__(self):
        self.route.plug("env")
        self.capabilities = MyCapabilities()

    @route(env_requires="pyjwt&redis")
    def create_jwt(self):
        return "jwt created"

    @route(env_requires="paypal|stripe")
    def process_payment(self):
        return "payment processed"

# Query with additional request capabilities
obj = MyAPI()
obj.route.node("create_jwt", env_capabilities="pyjwt")  # OK: pyjwt from request + redis from instance
obj.route.node("create_jwt")  # not_available: only redis from instance, missing pyjwt

# Dynamic capabilities via CapabilitiesSet
class PaymentCapabilities(CapabilitiesSet):
    def __init__(self, service):
        self._service = service

    @capability
    def stripe(self) -> bool:
        return self._service._stripe_configured

    @capability
    def paypal(self) -> bool:
        return self._service._paypal_configured

class PaymentService(RoutingClass):
    def __init__(self):
        self.route.plug("env")
        self._stripe_configured = True
        self._paypal_configured = False
        self.capabilities = PaymentCapabilities(self)

Rule syntax (on entry env_requires):

| : OR (system must have at least one)
& : AND (system must have all)
! : NOT (system must not have)
() : grouping

NOTE: Comma is NOT allowed in env_requires. Use | for OR, & for AND.

Comma in env_capabilities means the system has multiple capabilities.

Example: env_requires=”pyjwt&redis” means “system must have pyjwt AND redis” Example: env_requires=”paypal|stripe” means “system must have paypal OR stripe”

class genro_routes.plugins.env.EnvPlugin(router, **config)[source]

Bases: BasePlugin

Environment capability-based access control plugin.

Controls access to router entries based on system capabilities. Capabilities represent runtime features (installed modules, configured services, etc.) that may or may not be available.

Capability sources (combined with OR):

Instance capabilities: Declared on RoutingClass via CapabilitiesSet
Request capabilities: Passed via env_capabilities parameter
Accumulated capabilities: Inherited from parent RoutingClass instances

Rule syntax (on entry via env_requires):

| : OR (system must have at least one capability)
& : AND (system must have all capabilities)
! : NOT (system must not have capability)
() : grouping for complex expressions

plugin_code: “env” - used for registration and config prefix.

plugin_description: Human-readable description.

Example

Entry definition:

@route(env_requires="redis&pyjwt")  # requires both
def create_session(self): ...

@route(env_requires="stripe|paypal")  # requires one
def process_payment(self): ...

Dynamic capabilities via CapabilitiesSet:

class ServerCaps(CapabilitiesSet):
    @capability
    def redis(self) -> bool:
        return self._redis_client is not None

class MyService(RoutingClass):
    def __init__(self):
        self.route.plug("env")
        self.capabilities = ServerCaps()

Query with additional request capabilities:

# Adds pyjwt to instance capabilities for this request
router.node("create_session", env_capabilities="pyjwt")

Parameters:

router (Any)
config (Any)

plugin_code: str = 'env'

plugin_description: str = 'Environment capability-based access control plugin'

plugin_default_param: str | None = 'requires'

configure(*, requires='', enabled=True, _target='_all_', flags=None)[source]

Define capability requirements for this entry/router.

Parameters:

requires (str) – Boolean rule expression (e.g., “pyjwt&redis”, “paypal|stripe”). Use | for OR, & for AND. Comma is not allowed.
enabled (bool) – Whether the plugin is enabled (default True)
_target (str) – Internal - target bucket name
flags (str | None) – Internal - flag string

Raises:

ValueError – If requires contains comma (use | for OR instead).

Return type:

None

deny_reason(entry, **filters)[source]

Filter entries based on capability requirements.

Capabilities are accumulated from: 1. Router capabilities (walked from the router’s parent chain) 2. Request capabilities (capabilities parameter)

The combined set is checked against the entry’s env_requires.

Parameters:

entry (MethodEntry) – MethodEntry being checked.
**filters (Any) – May contain capabilities (from request), the only public env filter key.

Returns:

Access allowed (no reason to deny). “not_available”: Entry requires capabilities but none available,

or capabilities don’t match rule.

Return type:

””

name

class genro_routes.plugins.env.CapabilitiesSet[source]

Bases: object

Base class for dynamic capability sets.

Subclasses define capabilities as methods decorated with @capability. The class behaves like a set: supports in, len, and iteration.

Capabilities are evaluated dynamically on each access, allowing for runtime conditions (e.g., time of day, module availability, configuration).

Usage:

from genro_routes.plugins.env import CapabilitiesSet, capability

class ServerCapabilities(CapabilitiesSet):
    @capability
    def jwt(self) -> bool:
        return "pyjwt" in sys.modules

    @capability
    def send_mail(self) -> bool:
        hour = datetime.now().hour
        return 8 <= hour <= 20

caps = ServerCapabilities()
"jwt" in caps       # True if pyjwt is installed
list(caps)          # ["jwt", "send_mail"] (only active ones)
len(caps)           # number of active capabilities

Integration with RoutingClass:

class MyService(RoutingClass):
    def __init__(self):
        self.route.plug("env")
        self.capabilities = ServerCapabilities()

__iter__()[source]: Yield names of currently active capabilities.

__contains__(item)[source]

Check if a capability is currently active.

Parameters:: item (str)
Return type:: bool

__len__()[source]

Return the number of currently active capabilities.

Return type:: int

genro_routes.plugins.env.capability(func)[source]

Mark a method as a capability checker.

The decorated method should return a bool indicating whether the capability is currently active.

Usage:

class ServerCapabilities(CapabilitiesSet):
    @capability
    def jwt(self) -> bool:
        return "jwt" in sys.modules

    @capability
    def redis(self) -> bool:
        return self._redis_client is not None

Logging plugin for Genro Routes.

Wraps handler calls with configurable logging messages including timing.

Configuration

Accepted keys (router-level or per-handler):

enabled: Gate the plugin entirely (default True)
before: Log “start” message (default True)
after: Log “end” message with timing (default True)
log: Use logger.info() when available (default True)
print: Always use print() (default False)

Example:

from genro_routes import Router, RoutingClass, route

class MyService(RoutingClass):
    def __init__(self):
        self.route.plug("logging")

    @route()
    def hello(self):
        return "Hello!"

# Or configure per-handler:
@route(logging_after=False)
def fast_handler(self):
    return "fast"

class genro_routes.plugins.logging.LoggingPlugin(router, *, logger=None, **cfg)[source]

Bases: BasePlugin

Logging plugin with configurable start/end messages and timing.

Wraps handler invocations with optional “start” and “end” log messages, including execution timing in milliseconds.

Configuration options:

enabled: Enable/disable the plugin entirely (default True)
before: Log “{handler} start” before execution (default True)
after: Log “{handler} end (X ms)” after execution (default True)
log: Use logger.info() when handlers available (default True)
print: Always use print() instead of logger (default False)

Output sinks:

By default uses Python’s logging module (logging.getLogger("genro_routes")). Falls back to print() if no handlers are configured on the logger. Set print=True to always use print().

plugin_code: “logging” - used for registration and config prefix.

plugin_description: Human-readable description.

Example

Basic usage:

class MyService(RoutingClass):
    def __init__(self):
        self.route.plug("logging")

    @route()
    def hello(self):
        return "Hello!"

Per-handler configuration:

@route(logging_after=False)  # disable end message
def fast_handler(self):
    return "fast"

Runtime configuration:

svc.route.logging.configure(before=False)  # disable globally
svc.route.logging.configure(_target="slow_handler", after=True)

Parameters:: logger (Logger | None)

plugin_code: str = 'logging'

plugin_description: str = 'Logs handler calls with timing'

__init__(router, *, logger=None, **cfg)[source]

Parameters:: logger (Logger | None)

configure(enabled=True, before=True, after=True, log=True, print=False)[source]

Configure logging plugin options.

Parameters:

enabled (bool) – Enable/disable the plugin entirely.
before (bool) – Log “{handler} start” before execution.
after (bool) – Log “{handler} end (X ms)” after execution.
log (bool) – Use logger.info() when handlers available.
print (bool) – Always use print() instead of logger.

wrap_handler(route, entry, call_next)[source]

Wrap handler with start/end logging and timing.

Parameters:

entry (MethodEntry)
call_next (Callable)

ChannelPlugin - Channel-based endpoint filtering.

This plugin filters endpoint visibility based on the request channel (mcp, rest, bot_*, web, etc.). It works alongside AuthPlugin (who) and EnvPlugin (what’s available) to control endpoint access.

Usage:

from genro_routes import Router, RoutingClass, route

class MyAPI(RoutingClass):
    def __init__(self):
        self.route.plug("channel")
        self.route.channel.configure(channels="*")  # default: all channels

    @route(channel_channels="mcp")
    def mcp_only(self):
        return "only accessible from MCP channel"

    @route(channel_channels="mcp,bot_.*")
    def mcp_and_bots(self):
        return "MCP + any bot channel"

    @route()
    def everywhere(self):
        return "inherits * from router config"

obj = MyAPI()
obj.route.nodes(channel_channel="mcp")     # sees all three
obj.route.nodes(channel_channel="rest")    # sees only everywhere
obj.route.node("mcp_only", channel_channel="mcp")  # OK
obj.route.node("mcp_only", channel_channel="rest")  # NotAvailable

Channel patterns:

Comma-separated list of patterns: "mcp,bot_.*,rest"
Each pattern is matched with re.fullmatch against the request channel
"*" is a special wildcard that matches any channel
Empty string means no channels allowed (default closed)

class genro_routes.plugins.channel.ChannelPlugin(router, **config)[source]

Bases: BasePlugin

Channel-based endpoint filtering plugin.

Controls access to router entries based on the request channel. Channels represent where a request originates from (mcp, rest, bot_sourcerer, web, etc.).

Behavior:

Default closed: entry without channels configured returns "not_available"
Wildcard: channels="*" opens endpoint to all channels
Regex patterns: comma-separated, matched with re.fullmatch
Config inheritance: child routers inherit parent channel config

plugin_code: “channel” - used for registration and config prefix.

plugin_description: Human-readable description.

plugin_default_param: “channels” - enables shorthand syntax.

Parameters:

router (Any)
config (Any)

configure(*, channels='', enabled=True, _target='_all_', flags=None)[source]

Define allowed channels for this entry/router.

Parameters:

channels (str) – Comma-separated list of channel patterns. Each pattern is matched with re.fullmatch. Use "*" to allow all channels. Empty string means no channels allowed.
enabled (bool) – Whether the plugin is enabled (default True).
_target (str) – Internal - target bucket name.
flags (str | None) – Internal - flag string.

Return type:

None

deny_reason(entry, **filters)[source]

Filter entries based on channel.

Parameters:

entry (MethodEntry | RouterInterface) – MethodEntry or Router being checked.
**filters (Any) – May contain channel with request channel string.

Returns:

Access allowed (channel matches). “not_available”: Channel doesn’t match or not configured.

Return type:

””

name

Pydantic validation and response schema plugin for Genro Routes.

Validates handler inputs using Pydantic type hints and generates JSON Schema from return type annotations for bridge consumption (MCP, OpenAPI).

At registration time (on_decore): - Inspects parameter type hints and builds a Pydantic model for input validation. - Inspects return type annotation and generates a JSON Schema via TypeAdapter,

stored in entry.metadata["pydantic"]["response_schema"].

At call time (wrap_handler), validates annotated args/kwargs before calling the real handler.

Strict by default

Validation runs in strict mode: an argument must already have the annotated type. "12" for an int parameter is a ValidationError, not a conversion. Pydantic’s own strict-mode allowances still apply (an int is accepted for a float parameter).

Coercion is opt-in per call, with the reserved keyword _coerce=True passed to RouterNode.__call__. It applies to every parameter of that call: "12" becomes 12, "2026-09-01" becomes a date. _coerce is always consumed by the router and by this plugin’s wrapper, and never reaches the handler.

_coerce=True on a router without the pydantic plugin raises the exception mapped to the not_available error code. _coerce=True on an entry with no type hints (no model) or with validation disabled is silently ignored: there is nothing to convert.

A single parameter can opt out of strict mode regardless of _coerce, with Annotated[int, Field(strict=False)]: the per-field setting overrides the model-level strict flag.

Example:

from typing import TypedDict
from genro_routes import Router, RoutingClass, route

class UserResponse(TypedDict):
    id: int
    name: str

class MyService(RoutingClass):
    def __init__(self):
        self.route.plug("pydantic")

    @route()
    def get_user(self, user_id: int) -> UserResponse:
        return {"id": user_id, "name": "alice"}

svc = MyService()
svc.route.node("get_user")(user_id=123)  # OK, validated
svc.route.node("get_user")(user_id="123")  # ValidationError (strict)
svc.route.node("get_user")(user_id="123", _coerce=True)  # OK, converted
svc.route.node("get_user")(user_id="not_an_int")  # ValidationError

# Response schema available in metadata
entry = svc.route._entries["get_user"]
entry.metadata["pydantic"]["response_schema"]
# {"type": "object", "properties": {"id": ..., "name": ...}, ...}

Configuration:

# Disable validation for a specific handler
@route(pydantic_disabled=True)
def unvalidated_handler(self):
    pass

class genro_routes.plugins.pydantic.PydanticPlugin(router, **config)[source]

Bases: BasePlugin

Validate handler inputs and generate response schemas with Pydantic.

At registration time (on_decore), builds a Pydantic model from parameter type hints for input validation, and generates a JSON Schema from the return type annotation via TypeAdapter for bridge consumption.

Behavior:

Only annotated parameters are validated
Unannotated parameters pass through unchanged
Validation is strict: no type coercion unless the call passes _coerce=True, which converts every parameter of that call
Annotated[T, Field(strict=False)] opts a single parameter out of strict mode without _coerce
ValidationError is raised on invalid input
Return type annotations produce response_schema in metadata
Can be disabled per-handler via pydantic_disabled=True

Configuration options:

disabled: Skip validation for this handler/router (default False)

plugin_code: “pydantic” - used for registration and config prefix.

plugin_description: Human-readable description.

Example

Basic usage:

class MyService(RoutingClass):
    def __init__(self):
        self.route.plug("pydantic")

    @route()
    def get_user(self, user_id: int) -> dict[str, int]:
        return {"id": user_id}

svc = MyService()
svc.route.node("get_user")(user_id=123)           # OK
svc.route.node("get_user")(user_id="123")         # ValidationError
svc.route.node("get_user")(user_id="123", _coerce=True)  # OK -> 123
svc.route.node("get_user")(user_id="not_an_int")  # ValidationError

# Response schema in metadata
svc.route._entries["get_user"].metadata["pydantic"]["response_schema"]

Disable validation:

@route(pydantic_disabled=True)
def unvalidated(self, data):
    return data  # no validation

Parameters:: config (Any)

plugin_code: str = 'pydantic'

plugin_description: str = 'Validates inputs and generates response schemas using Pydantic'

__init__(router, **config)[source]

Parameters:: config (Any)

configure(disabled=False)[source]

Configure pydantic plugin options.

Parameters:: disabled (bool) – If True, skip validation for this handler/router.

on_decore(route, func, entry)[source]

Build Pydantic model from handler type hints and generate response schema.

Parameters:

route (Router)
func (Callable)
entry (MethodEntry)

Return type:

None

wrap_handler(route, entry, call_next)[source]

Validate annotated parameters with the cached Pydantic model before calling.

The wrapper is installed even for an entry without a model, because it owns the reserved _coerce keyword: it must consume it in every case so the handler never sees it. With a model, _coerce selects lax validation for this call; without one there is nothing to convert and the keyword is simply dropped.

Parameters:

route (Router)
entry (MethodEntry)
call_next (Callable)

get_model(entry)[source]

Return the Pydantic model for this handler if not disabled.

Parameters:: entry (MethodEntry) – The MethodEntry to get the model for.
Return type:: tuple[str, Any] | None
Returns:: Tuple of (“pydantic_model”, model_class) if available, else None.

name

entry_metadata(router, entry)[source]

Return pydantic metadata for introspection.

Parameters:

router (Any)
entry (MethodEntry)

Return type:

dict[str, Any]

Plugin API Reference

External plugin API

PydanticPlugin: strict validation and _coerce

Auto-Generated Plugin API

Configuration

Strict by default

PydanticPlugin: strict validation and `_coerce`