Communication Subsystem

The platform communication subsystem is a three-layer pipeline that delivers async events from bundles to connected clients. This is the same delivery path used by the default chat UI, bundle widgets, and custom bundle main views. It is transport-agnostic at the event level — the same bundle code can emit progress, deltas, and completion events without caring whether the connected client consumes them over SSE or Socket.IO.

Bundle REST operations and bundle-served MCP endpoints are sibling proc surfaces that sit next to this pipeline, not inside the event envelope itself. They can still emit communicator events during a request, but the underlying HTTP request/response or MCP exchange remains a separate surface.

ChatCommunicator — Bundle Producer API

Every bundle workflow receives a ChatCommunicator instance. It wraps Redis Pub/Sub with a typed API for all outbound async events. The same instance works whether the bundle runs in-process or inside an isolated execution runtime (Docker / Fargate sandbox) — the iso executor routes events to the supervisor over a Unix socket, which then publishes to Redis. Bundle-defined chat UIs, widgets, and custom frontends see the same stream surface in both cases.

# Standard streaming lifecycle
await communicator.start()                   # chat_start event
await communicator.step("Searching…")       # chat_step — visible progress
await communicator.delta("answer", chunk)   # chat_delta — streaming text
await communicator.event(
    event_type="chat.compaction",
    data={"status": "completed"},
)                                               # chat_compaction — context compaction status
await communicator.complete()                # chat_complete

# Delta markers: answer · thinking · canvas · timeline_text · subsystem
await communicator.delta("thinking", reasoning_chunk)
await communicator.delta("canvas", json_payload)

# Custom service event — broadcast to all session subscribers
await communicator.event(
    event_type="chat.service",
    data={"key": "status.update", "value": "processing"},
)

# P2P — deliver only to a specific connected client (target_sid)
await communicator.event(
    event_type="chat.service",
    data={"key": "private", "value": result},
    target_sid=sid,                           # omit for broadcast
)

# Error
await communicator.error("Something went wrong")

Outbound Firewall (per bundle)

Bundles can attach an IEventFilter to the workflow to suppress or gate events before they reach Redis. The filter sees the caller's user_type, user_id, and full event metadata. Filters are fail-open: an exception allows the event through.

from kdcube_ai_app.apps.chat.sdk.comm import IEventFilter

class MyBundleFilter(IEventFilter):
    def allow_event(self, user_type, user_id, event) -> bool:
        # hide internal step events from non-privileged users
        if user_type != "privileged":
            if event.get("type") == "chat.step" and event.get("broadcast"):
                return False
        return True

# Wire at entrypoint — passed to workflow factory via event_filter param
workflow = MyWorkflow(..., event_filter=MyBundleFilter())

See comm-system.md and README-comm.md for full detail.

Streaming vs Request-Scoped Surfaces

The communication pipeline carries asynchronous bundle events. It does not mean every bundle interface is itself a chat stream. KDCube currently exposes several adjacent surfaces, each with a different contract.

A bundle operation or MCP request can still produce live UI updates when the bundle code explicitly emits communicator events. In that case the streaming delivery reuses the same SSE or Socket.IO session identified by auth state and optional KDC-Stream-ID.

Bundle-served MCP transport contract

For MCP, there are three separate responsibilities:

• Proc: resolve the bundle route, forward the original HTTP request, and dispatch into the returned FastMCP or ASGI subapp
• Bundle: decide whether the request is public or authenticated, read whatever headers/cookies it wants, and verify them itself
• Client: call the explicit MCP route and send the auth material that the bundle contract requires

That means the MCP request is not using the normal browser/user-flow API auth contract. AUTH.ID_TOKEN_HEADER_NAME is not the MCP auth header. The bundle may choose a completely different header such as X-Versatile-Preferences-MCP-Token, or a bearer token, cookie, HMAC signature, or custom JWT scheme.

Typical authenticated MCP flow:

1. bundle props define a non-secret contract such as the header name
2. bundle secrets store the verification material such as a shared token
3. the MCP client calls /api/integrations/bundles/{tenant}/{project}/{bundle_id}/mcp/{alias}
4. proc forwards headers and body unchanged into the bundle MCP app
5. the bundle verifies the request and either raises 401 or returns the FastMCP app

Typical public MCP flow:

1. the client calls /api/integrations/bundles/{tenant}/{project}/{bundle_id}/public/mcp/{alias}
2. proc forwards the request into the bundle MCP app
3. the bundle either accepts it as public or still applies its own auth logic if that is the chosen contract

Bundle-authenticated public API hooks

Public bundle APIs now have two different auth ownership shapes:

• Proc-owned: public_auth="none" or public_auth={"mode":"header_secret", ...}
• Bundle-owned: public_auth="bundle"

For public_auth="bundle", there are three separate responsibilities:

• Proc: resolve the route, parse the request, and invoke the bundle method
• Bundle: accept request: Request, read the inbound headers/body, and decide whether the hook is valid
• Client / webhook provider: call the explicit public operations route and send the auth material the bundle contract requires

Typical bundle-authenticated public hook flow:

1. bundle props define a non-secret contract such as the header name
2. bundle secrets store the verification material such as a shared token
3. the caller sends POST /api/integrations/bundles/{tenant}/{project}/{bundle_id}/public/{alias}
4. proc forwards the request into the bundle method
5. the bundle verifies the request and either raises 401 / 403 or returns normally