5. Gateway & Transports

How transport config on @component, the Gateway, and Transport adapters turn runnables into REST endpoints, MCP tools, and CLI commands.

Trace a Request: From HTTP to @runnable

Before explaining the architecture, let’s trace what actually happens when an HTTP request hits a SignalPy-generated endpoint.

Suppose you have this component:

@component("splunk", version="1.0",
           rest={"prefix": "/splunk", "version": "v1"})
class SplunkApp:
    @runnable("query", params=QueryParams, description="Run a Splunk query")
    async def query(self, params):
        return {"results": ["event1", "event2"]}

And a client sends: POST /api/v1/splunk/query {"spl": "index=main"}

Here is every step that executes:

Step 1: @runnable stores metadata (decoration time)

The @runnable decorator creates a RunnableDef and appends it to the class metadata. Nothing runs. No handler is registered yet.

# component.py — what @runnable actually does:
def runnable(name, *, params, description, ...):
    def decorator(fn):
        fn.__runnable__ = RunnableDef(
            name=name,                  # "query"
            fn=fn,                      # SplunkApp.query
            params_model=QueryParams,   # for validation
            description=description,
            is_async=True,              # detected from async def
        )
        return fn
    return decorator

At this point, SplunkApp is just a class with metadata attached. No kernel, no routes.

Step 2: Boot activates the component

When kernel.boot() activates SplunkApp, the kernel builds its ReactiveRuntime and runs lifecycle hooks. The runnable schemas are discoverable via kernel.runnables().

Step 3: Transport discovers runnables

The RESTTransport uses kernel.runnables_by_component(transport="rest") to discover all runnables whose component declares rest={...} config. Each runnable schema includes a handler reference — the bound method on the component instance.

Step 4: Transport generates a FastAPI route (rest.py)

The RESTTransport (or mount_rest()) generates routes that call the handler directly:

# rest.py — the actual route handler:
def _make_handler(schema):
    async def _handler(request: Request):
        body = await request.json()                       # {"spl": "index=main"}
        validated = schema.params_model.model_validate(body)
        result = await schema.handler(validated)           # direct call
        return JSONResponse({"ok": True, "data": result})
    return _handler

router.add_api_route("/query", _make_handler(schema), methods=["POST"])

FastAPI now has: POST /api/v1/splunk/query -> _handler

The transport calls schema.handler directly. No bus dispatch, no string-based routing. The schema is the bridge.

Step 5: HTTP request arrives

Client: POST /api/v1/splunk/query  {"spl": "index=main"}
         |
         v
FastAPI matches route -> _handler(request)              # rest.py
         |
         v
body = await request.json()                            # {"spl": "index=main"}
         |
         v
validated = QueryParams.model_validate(body)           # params validation
         |
         v
result = await schema.handler(validated)               # direct method call
         |
         v
return JSONResponse({"ok": True, "data": {"results": ["event1", "event2"]}})

That’s the full path. Four hops: FastAPI route -> param validation -> schema handler -> your method. The component never imported FastAPI. FastAPI never imported SignalPy.

Schema is the bridge

Every adapter (REST, MCP, CLI) generates the same thing: a thin wrapper that validates params and calls schema.handler(validated) directly. The runnable schema is the only connection between “presentation” and “capability.”

The Three-Layer Pipeline

Now that you’ve seen the trace, here’s the architecture:

Components don’t build REST routes or CLI commands. They declare what they want to expose, and the kernel handles the rest through three layers:

Layer 1: Component declares     @component("splunk", rest={"prefix": "/splunk"})
                                @runnable("query", ...)
                                        |
                                        v
Layer 2: Kernel indexes         kernel.runnables() / kernel.runnables_by_component()
                                collect all runnable schemas
                                        |
                                        v
Layer 3: Transport renders      RESTTransport reads runnables for "rest",
                                generates FastAPI routes
                                MCPTransport reads runnables for "mcp",
                                generates MCP tool definitions

The component never sees the transport. The transport discovers runnables through the kernel’s schema API.

Layer 1: Transport Config on @component

Transport configuration is declared directly on @component:

@component("splunk", version="1.0",
1           rest={"prefix": "/splunk", "version": "v1"},
2           mcp={"name": "splunk-tools"})
@requires(config=IConfig)
class SplunkApp:

    @runnable("query", params=QueryParams, description="Run a Splunk query")
    async def query(self, params):
        return {"results": [...]}

    @runnable("_reindex", params=BaseModel, transports=["native"], description="Reindex")
3    async def reindex(self, params):
        return {"ok": True}

1

“Expose my runnables as REST at /api/v1/splunk/*”

2

“Also expose them as MCP tools under the group splunk-tools”

3

transports=["native"] – callable within the kernel only, excluded from external surfaces

Transport config keys on @component:

Key	Purpose	Example
rest	REST adapter config	{"prefix": "/splunk", "version": "v1"}
mcp	MCP adapter config	{"name": "splunk-tools"}
cli	CLI adapter config	{"prefix": "splunk"}

Each transport key is a dict with adapter-specific fields like prefix, version, name, auth, include, exclude.

No transport config? Fallback: all runnables (except transports=["native"]) are exposed on every transport. Convenient for development; explicit config recommended for production.

Layer 2: The Kernel Schema API

The kernel indexes all runnable schemas at boot. Transport adapters query them:

# All runnables across all components
all_schemas = kernel.runnables()

# Runnables for a specific transport (components that declared rest={...})
rest_schemas = kernel.runnables_by_component(transport="rest")

Each schema includes the handler reference, params model, description, and transport restrictions. The kernel re-indexes when components are hot-added or hot-removed.

Layer 3: Transport Adapters — Rendering

Each transport adapter is a component that queries kernel.runnables_by_component() at activation and generates the actual endpoints.

REST (src/signalpy/adapters/rest.py)

@component("rest-transport")
class RESTTransport:

    @lifecycle.activate
    def activate(self):
        self.app = FastAPI(title="Microkernel API")

1        by_component = self.kernel.runnables_by_component(transport="rest")

        for comp_name, schemas in by_component.items():
2            transport_cfg = schemas[0].component_meta.rest
            prefix = transport_cfg.get("prefix", f"/{comp_name}")
            version = transport_cfg.get("version", "v1")
            router = APIRouter(prefix=f"/api/{version}{prefix}")

            for schema in schemas:
3                router.add_api_route(
                    f"/{schema.name}",
                    make_handler(schema),
                    methods=["POST"],
                )

            self.app.include_router(router)

1

Ask the kernel for all runnables on components that declared rest={...}

2

Read the component’s REST transport config

3

Each schema becomes a POST route that calls schema.handler directly

For the Splunk example, this generates:

POST /api/v1/splunk/query     -> schema.handler(validated)
POST /api/v1/spell/check      -> schema.handler(validated)
GET  /health                   -> {"status": "healthy"}

The handler is simple – it deserializes the request body, validates params, and calls the schema handler directly:

async def handler(request: Request):
    body = await request.json()
    validated = schema.params_model.model_validate(body)
    result = await schema.handler(validated)
    return JSONResponse({"ok": True, "data": result})

MCP (src/signalpy/adapters/mcp.py)

Same pattern – queries runnables for the “mcp” transport:

by_component = kernel.runnables_by_component(transport="mcp")
for comp_name, schemas in by_component.items():
    for schema in schemas:
        tools.append({
            "name": f"{comp_name}.{schema.name}",
            "description": schema.description,
            "input_schema": schema.params_model.model_json_schema(),
        })

CLI (src/signalpy/adapters/cli.py)

Same pattern – queries runnables for the “cli” transport:

by_component = kernel.runnables_by_component(transport="cli")
for comp_name, schemas in by_component.items():
    group = click.Group(name=comp_name)
    for schema in schemas:
        group.add_command(make_click_command(schema))

The Boot Sequence

1. kernel.discover([ConfigProvider, ..., SplunkApp, RESTTransport])
2. kernel.boot()
   a. Activate ConfigProvider, LoggingProvider, ...
   b. Activate SplunkApp
      -> runnable schemas indexed in kernel
   c. Activate RESTTransport
      -> queries kernel.runnables_by_component(transport="rest")
      -> generates FastAPI routes from schemas

Transport adapters activate after application components, so all runnable schemas are available when the transport queries them.

Two Modes: Container vs Library

SignalPy supports both deployment patterns:

Container mode — SignalPy is the runtime

SignalPy boots, creates and manages the FastAPI app (or MCP server) as a component. Good for microservices where SignalPy IS the application.

from signalpy.kernel import Kernel
from signalpy.providers.config import ConfigProvider
from signalpy.providers.logging_provider import LoggingProvider
from signalpy.providers.gateway import APIGateway
from signalpy.adapters.rest import RESTTransport

kernel = Kernel()
kernel.discover([
    ConfigProvider, LoggingProvider,
    SplunkApp,        # rest={"prefix": "/splunk", "version": "v1"}
    SpellChecker,     # rest={"prefix": "/spell"}
    RESTTransport,    # creates and owns the FastAPI app
])
await kernel.boot()

# Get the app SignalPy created
1app = kernel.registry.require("IRestAPI").get_app()
uvicorn.run(app, port=8000)

1

kernel.registry.require(C) returns the active service that satisfies contract C (raises if none). It’s the script-level escape hatch for reaching a service when you don’t have a component context. Inside a component, you’d @requires(...) and read self.rt.X instead. (require_optional(C) returns None rather than raising.)

Library mode — mount into your existing app

You already have a FastAPI app or FastMCP server. SignalPy mounts its runnables into it. Good for adding kernel-managed services to existing applications.

from fastapi import FastAPI
from signalpy.kernel import Kernel
1from signalpy.adapters.rest import mount_rest

# Your existing app
app = FastAPI(title="My App")

@app.get("/my-custom-endpoint")
async def custom():
    return {"custom": True}

# Boot the kernel (no RESTTransport needed)
kernel = Kernel()
kernel.discover([ConfigProvider, LoggingProvider, SplunkApp])
await kernel.boot()

# Mount kernel runnables as routes on YOUR app
2mount_rest(app, kernel)

# Now your app has both custom routes AND kernel routes:
#   GET  /my-custom-endpoint       -> your code
#   POST /api/v1/splunk/query      -> schema.handler(validated)
#   GET  /health                   -> kernel health

1

mount_rest is a function, not a component. No Transport component needed.

2

Reads runnable schemas from the kernel directly.

Same for MCP:

from fastmcp import FastMCP
from signalpy.adapters.mcp import mount_mcp

server = FastMCP("my-server")

# Your own tools
@server.tool()
def my_custom_tool(query: str) -> str:
    return "custom result"

# Mount kernel runnables as MCP tools
mount_mcp(server, kernel)

# Now the server has both your tools AND kernel runnables as tools

When to use which

Mode	When	What you discover
Container	SignalPy IS the app (microservice)	RESTTransport / MCPTransport components
Library	Adding to existing FastAPI/FastMCP	Call mount_rest() / mount_mcp() functions

Both read from the same gateway surface. Both generate the same routes/tools. The difference is who owns the framework instance.

What Happens Without Transports?

Nothing breaks. Components still work – you call their methods directly through @requires injection:

@component("my-app")
@requires(splunk=ISplunkApp)
class MyApp:
    async def do_work(self):
        result = await self.rt.splunk.query(QueryParams(spl="index=main"))

Transport config on @component is just metadata. If no REST transport is discovered, no routes are generated. The component works identically either way.

This is the separation: runnables are capabilities. Transport config is presentation. Components work without presentation.

Writing a Custom Transport

As a component (container mode):

@component("grpc-transport")
class GRPCTransport:
    @lifecycle.activate
    def activate(self):
        by_component = self.kernel.runnables_by_component(transport="grpc")
        for comp_name, schemas in by_component.items():
            for schema in schemas:
                self._register_grpc_method(schema)

As a mount function (library mode):

def mount_grpc(server, kernel):
    by_component = kernel.runnables_by_component(transport="grpc")
    for comp_name, schemas in by_component.items():
        for schema in schemas:
            server.add_method(
                f"{comp_name}.{schema.name}",
                lambda args, s=schema: s.handler(s.params_model.model_validate(args)),
            )

Components declare grpc={...} on @component and either approach picks them up. No kernel changes needed.