5. Gateway & Transports

How @api, 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. Every file path and line number references the real source.

Suppose you have this component:

@component("splunk", version="1.0")
@api("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, with the actual code:

Step 1: @runnable stores metadata (decoration time)

component.py:329 — 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 bus, no routes.

Step 2: Boot registers a bus handler (kernel/__init__.py:336-338)

When kernel.boot() activates SplunkApp, _register_component_bus() reads the metadata and registers a handler on the bus:

# kernel/__init__.py:336-338
for rd in ci.meta.runnables:
    handler_name = f"{ci.name}.{rd.name}"    # "splunk.query"
    self.bus.register_handler(
        handler_name,
        self._make_bus_handler(rd, ci.instance, ci),
    )

_make_bus_handler (kernel/__init__.py:272-317) creates a closure that:

1. Validates params with the Pydantic model
2. Checks auth if requires_action or requires_role is set
3. Calls rd.fn(instance, validated_params) — the actual query() method

Now bus._handlers["splunk.query"] is a callable. The bus doesn’t know about REST or MCP. It’s just a dict of {name: async_function}.

Step 3: Gateway builds a surface (gateway.py:82-135)

The APIGateway component activates after SplunkApp. It scans bus.handlers and matches each handler to its component’s @api declarations:

# gateway.py:93-97 — for each bus handler:
handler_name = "splunk.query"
component_qn = "splunk"
runnable_name = "query"

# gateway.py:113 — for each @api declaration on that component:
for api_def in meta.apis:          # APIDef(transport="rest", prefix="/splunk", version="v1")
    surface = self._get_surface("rest")
    surface.entries.append(APIEntry(
        runnable_name="splunk.query",   # the bus handler name — this is the key
        short_name="query",
        group="splunk",                 # from prefix="/splunk"
    ))

The gateway now holds: surfaces["rest"] contains an APIEntry with runnable_name="splunk.query". This string is the bridge between the gateway and the bus.

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

The RESTTransport (or mount_rest()) reads the surface and generates routes. The core is this closure in rest.py:_add_route():

# rest.py — the actual route handler:
def _make_handler(name="splunk.query", b=bus):
    async def _handler(request: Request):
        body = await request.json()                  # {"spl": "index=main"}
        result = await b.invoke(name, body)          # ← THE BRIDGE
        return JSONResponse({"ok": True, "data": result})
    return _handler

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

FastAPI now has: POST /api/v1/splunk/query → _handler

The bridge is one line: await bus.invoke("splunk.query", body). The route handler doesn’t know about SplunkApp. It only knows the bus handler name.

Step 5: HTTP request arrives

Client: POST /api/v1/splunk/query  {"spl": "index=main"}
         │
         ▼
FastAPI matches route → _handler(request)              # rest.py
         │
         ▼
body = await request.json()                            # {"spl": "index=main"}
         │
         ▼
result = await bus.invoke("splunk.query", body)        # rest.py → bus.py:74
         │
         ▼
handler = bus._handlers["splunk.query"]                # bus.py:83-85
result = await handler({"spl": "index=main"})          #   ↓
         │                                             #   ↓
         ▼                                             #   ↓
_make_bus_handler closure:                             # kernel/__init__.py:296
  validated = QueryParams.model_validate(body)         #   params validation
  result = await SplunkApp.query(instance, validated)  #   actual method call
         │
         ▼
return JSONResponse({"ok": True, "data": {"results": ["event1", "event2"]}})

That’s the full path. Five hops: FastAPI route → bus.invoke → bus handler → param validation → your method. The component never imported FastAPI. FastAPI never imported SignalPy. The bus handler name ("splunk.query") is the only connection between them.

The bus is the bridge

Every adapter (REST, MCP, CLI) generates the same thing: a thin wrapper that calls bus.invoke(runnable_name, params). The bus handler name is a string. That’s the entire interface 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     @api("rest", prefix="/splunk")
                                @runnable("query", ...)
                                        │
                                        ▼
Layer 2: Gateway composes       APIGateway scans all components,
                                builds one APISurface per transport
                                        │
                                        ▼
Layer 3: Transport renders      RESTTransport reads the "rest" surface,
                                generates FastAPI routes
                                MCPTransport reads the "mcp" surface,
                                generates MCP tool definitions

The component never sees the transport. The transport never sees the component. The gateway is the meeting point.

Layer 1: @api — What the Component Declares

@api attaches metadata — it does nothing at runtime:

0@component("splunk", version="1.0", depends=["config"])
@requires(config="IConfig")
1@api("rest", prefix="/splunk", version="v1")
2@api("mcp", name="splunk-tools")
class SplunkApp:

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

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

0

depends=["config"] — explicit activation-order constraint. Use this when you need a component to be active before yours but you don’t have a @requires on it (e.g., you read its env-var side effects). With @requires(config=IConfig) already declared, this depends= is technically redundant; we show it to demonstrate the parameter.

1

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

2

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

3

internal=True — on the bus but excluded from all API surfaces

You can stack multiple @api decorators. Each one targets a different transport. The APIDef dataclass stores the declaration:

Field	Purpose	Example
transport	Which adapter picks this up	"rest", "mcp", "cli"
prefix	URL/command prefix	"/splunk" → /api/v1/splunk/query
name	Group name (MCP/CLI)	"splunk-tools"
version	API version	"v1" → /api/v1/...
auth	Require auth for this surface	True / False
include	Only expose these runnables	["query", "status"]
exclude	Exclude these runnables	["debug"]

No @api? Fallback: all non-internal runnables are exposed on every transport. This is convenient for development but explicit @api is recommended for production.

Layer 2: The Gateway — Composition

APIGateway is a component (src/signalpy/providers/gateway.py). At boot, it scans all active components and builds APISurface objects — one per transport type.

# Simplified from gateway.py:
def _rebuild(self):
    for handler_name in self._bus.handlers:          # "splunk.query", "splunk._reindex"
        component_name, runnable_name = handler_name.rsplit(".", 1)
        meta = self._find_meta(component_name)

        for api_def in meta.apis:                    # @api("rest", ...), @api("mcp", ...)
            # Skip internal runnables
            if runnable.internal:
                continue
            # Apply include/exclude filters
            if api_def.include and runnable_name not in api_def.include:
                continue

            surface = self._get_surface(api_def.transport)   # "rest" → APISurface
            surface.entries.append(APIEntry(
                runnable_name=handler_name,          # "splunk.query"
                short_name=runnable_name,            # "query"
                group=api_def.prefix or meta.factory_name,   # "splunk"
                ...
            ))

After scanning, the gateway holds:

surfaces = {
    "rest": APISurface(
        transport="rest",
        version="v1",
        entries=[
            APIEntry(runnable_name="splunk.query", short_name="query", group="splunk"),
            APIEntry(runnable_name="checker.check", short_name="check", group="spell"),
        ]
    ),
    "mcp": APISurface(
        transport="mcp",
        entries=[
            APIEntry(runnable_name="splunk.query", short_name="query", group="splunk-tools"),
        ]
    ),
}

The gateway rebuilds when components are hot-added or hot-removed. This is called automatically by kernel.hot_add() and kernel.hot_remove().

Layer 3: Transport Adapters — Rendering

Each transport adapter is a component that @requires(gateway="IGateway"). At activation, it reads its surface from the gateway and generates the actual endpoints.

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

@component("rest-transport", depends=["gateway"])
@requires(config="IConfig", gateway="IGateway")
class RESTTransport:

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

1        surface = rt.gateway.get_surface("rest")

2        for group_name, entries in surface.groups().items():
            router = APIRouter(prefix=f"/api/{surface.version}/{group_name}")

            for entry in entries:
3                router.add_api_route(
                    f"/{entry.short_name}",
                    make_handler(entry.runnable_name, rt.bus),
                    methods=["POST"],
                )

            self.app.include_router(router)

1

Ask the gateway for the “rest” surface

2

Group entries by prefix — each group becomes a FastAPI router

3

Each entry becomes a POST route that calls bus.invoke(runnable_name, body)

rt.gateway and rt.bus

rt.gateway is just a normal @requires(gateway="IGateway") injection (you’d reach it as self.rt.gateway with the modern def activate(self): style). rt.bus, on the other hand, is the kernel bus itself — always present on every runtime, no @requires needed. Use it when you need to invoke runnables by name from non-component code (transports, tests, scripts).

For the Splunk example, this generates:

POST /api/v1/splunk/query     → bus.invoke("splunk.query", body)
POST /api/v1/spell/check      → bus.invoke("checker.check", body)
GET  /health                   → {"status": "healthy"}
GET  /api/kernel/status        → surface summary

The handler is simple — it deserializes the request body, calls bus.invoke(), and wraps the result in JSON:

async def handler(request: Request):
    body = await request.json()
    result = await bus.invoke("splunk.query", body)
    return JSONResponse({"ok": True, "data": result})

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

Same pattern — reads the “mcp” surface, generates tool definitions:

surface = rt.gateway.get_surface("mcp")
for entry in surface.entries:
    tools.append({
        "name": f"{entry.group}.{entry.short_name}",
        "description": entry.description,
        "input_schema": entry.params_model.model_json_schema(),
    })

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

Same pattern — reads the “cli” surface, generates Click commands:

surface = rt.gateway.get_surface("cli")
for group_name, entries in surface.groups().items():
    group = click.Group(name=group_name)
    for entry in entries:
        group.add_command(make_click_command(entry))

The Boot Sequence

1. kernel.discover([ConfigProvider, ..., SplunkApp, RESTTransport])
2. kernel.boot()
   a. Activate ConfigProvider, LoggingProvider, ...
   b. Activate SplunkApp
      → registers bus handler "splunk.query"
   c. Activate APIGateway
      → scans bus handlers, builds APISurfaces
   d. Activate RESTTransport
      → reads gateway.get_surface("rest")
      → generates FastAPI routes
   e. Gateway rebuild (phase 2)
      → re-scans with full component metadata
      → transports re-activate with complete surfaces

Phase 2 matters: the gateway first builds with just bus handler names (step c), then rebuilds with full metadata after all components are active (step e). This handles the ordering problem — transports need gateway, which needs all apps to be registered first.

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,
    APIGateway,
    SplunkApp,        # @api("rest", prefix="/splunk", version="v1")
    SpellChecker,     # @api("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, APIGateway, 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      → bus.invoke("splunk.query", body)
#   GET  /health                   → kernel health

1

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

2

Reads the gateway surface (if available) or falls back to all bus handlers.

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 Gateway or Transports?

Nothing breaks. Bus calls still work:

result = await kernel.bus.invoke("splunk.query", {"spl": "index=main"})

The @api declarations are just metadata. If no gateway is discovered, they’re ignored. If no REST transport is discovered, no routes are generated. The component works identically either way — it’s still a bus handler.

The mount_rest() / mount_mcp() functions also work without a gateway — they fall back to exposing all bus handlers directly.

This is the separation: runnables are capabilities. @api is presentation. The bus works without presentation.

Writing a Custom Transport

As a component (container mode):

@component("grpc-transport", depends=["gateway"])
@requires(gateway="IGateway")
class GRPCTransport:
    @lifecycle.activate
    def activate(self):
        surface = self.rt.gateway.get_surface("grpc")
        if surface:
            for entry in surface.entries:
                self._register_grpc_method(entry)

As a mount function (library mode):

def mount_grpc(server, kernel):
    gateway = kernel.registry.require_optional("IGateway")
    surface = gateway.get_surface("grpc") if gateway else None
    for entry in (surface.entries if surface else []):
        server.add_method(entry.runnable_name, lambda args: kernel.bus.invoke(...))

Components declare @api("grpc", ...) and either approach picks them up. No kernel changes. No gateway changes.