Middleware Components

The Kailash middleware layer provides enterprise-grade components for building production applications with real-time communication, session management, and AI integration capabilities.

Overview

The middleware architecture consists of composable components that work together to provide a complete solution for frontend-backend communication in workflow-based applications.

Core Components

Agent-UI Middleware

The AgentUIMiddleware serves as the central orchestration hub for frontend communication, providing session management, dynamic workflow creation, and execution monitoring.

Key Features:

Multi-tenant session isolation
Dynamic workflow creation from JSON configurations
Real-time execution monitoring
Automatic session cleanup
Database persistence support

Usage Example:

from kailash.middleware import AgentUIMiddleware

agent_ui = AgentUIMiddleware(
    max_sessions=1000,
    session_timeout_minutes=60,
    enable_persistence=True
)

# Create session
session_id = await agent_ui.create_session("user123")

# Create dynamic workflow
workflow_config = {
    "name": "data_pipeline",
    "nodes": [...],
    "connections": [...]
}

workflow_id = await agent_ui.create_dynamic_workflow(
    session_id, workflow_config
)

API Gateway

The APIGateway provides RESTful API endpoints with authentication, CORS, and automatic OpenAPI documentation generation.

Key Features:

RESTful API endpoints for workflow management
JWT authentication integration
CORS configuration
Automatic OpenAPI/Swagger documentation
Health monitoring endpoints

Usage Example:

from kailash.middleware import create_gateway

gateway = create_gateway(
    title="My Production API",
    cors_origins=["https://myapp.com"],
    enable_docs=True,
    enable_auth=True
)

# Gateway provides automatic endpoints:
# POST /api/sessions - Create session
# POST /api/workflows - Create workflow
# POST /api/executions - Execute workflow
# GET /health - Health check
# GET /docs - API documentation

Real-time Middleware

Provides real-time communication capabilities including WebSocket and Server-Sent Events for live workflow updates.

Key Features:

WebSocket bi-directional communication
Server-Sent Events (SSE) streaming
Event filtering and subscription management
Webhook delivery for external integrations
Automatic reconnection handling

Usage Example:

from kailash.middleware import RealtimeMiddleware

realtime = RealtimeMiddleware(agent_ui)

# Subscribe to events
async def handle_events(event):
    print(f"Event: {event.type} - {event.data}")

await realtime.event_stream.subscribe(
    "my_listener", handle_events
)

Event System

Event Stream

Comprehensive event management system for handling workflow and system events.

Event Types

Event Filtering

Event Classes

Authentication & Security

JWT Authentication Manager

Bases: object

Enterprise JWT Authentication Manager.

Provides comprehensive JWT token management with security best practices: - Support for both HS256 (default) and RSA algorithms - RSA key pair generation and rotation (when using RSA) - Refresh token management - Token revocation via a pluggable revocation store - Comprehensive audit logging - Rate limiting protection

This consolidates both JWTAuthManager and KailashJWTAuthManager functionality.

Token revocation in multi-worker deployments

Token revocation is backed by a TokenRevocationStore. By default (revocation_store omitted, config.enable_blacklist=True) an in-memory InMemoryTokenRevocationStore is used, which is process-local: a token revoked through one worker is NOT rejected by other workers. For any multi-worker / multi-pod deployment supply a SHARED store (Redis, database, distributed cache) implementing TokenRevocationStore via the revocation_store constructor argument so revocation propagates to every worker that shares it (issue #1356).

Known process-local state (NOT covered by revocation_store): refresh-token tracking (refresh_access_token / revoke_refresh_token / revoke_all_user_tokens) and rate-limit accounting are also per-instance in-memory and behave per-worker. Shared-backend coverage for those is tracked separately; only access-token revocation propagates through the store.

Initialize JWT Auth Manager.

Parameters:

config (JWTConfig | None) – JWTConfig object with full configuration
secret_key (str | None) – Secret key for HS256 (overrides config)
algorithm (str | None) – Algorithm to use (overrides config)
use_rsa (bool | None) – Whether to use RSA (overrides config)
revocation_store (TokenRevocationStore | None) – Backend that records and checks token revocation. When config.enable_blacklist is True and this is omitted, a process-local InMemoryTokenRevocationStore is used — which means a token revoked on one worker is NOT rejected by other workers. Supply a SHARED store (Redis, database, distributed cache) implementing TokenRevocationStore so revocation propagates across every worker that shares it (issue #1356). Ignored when config.enable_blacklist is False.
**kwargs – Additional config parameters

create_access_token(user_id: str, tenant_id: str | None = None, session_id: str | None = None, permissions: List[str] | None = None, roles: List[str] | None = None, **kwargs) → str[source]

Create JWT access token.

Parameters:

user_id (str)
tenant_id (str | None)
session_id (str | None)
permissions (List[str] | None)
roles (List[str] | None)

Return type:

str

create_refresh_token(user_id: str, tenant_id: str | None = None, session_id: str | None = None, **kwargs) → str[source]

Create JWT refresh token.

Parameters:

user_id (str)
tenant_id (str | None)
session_id (str | None)

Return type:

str

create_token_pair(user_id: str, tenant_id: str | None = None, session_id: str | None = None, permissions: List[str] | None = None, roles: List[str] | None = None, **kwargs) → TokenPair[source]

Create access and refresh token pair.

Parameters:

user_id (str)
tenant_id (str | None)
session_id (str | None)
permissions (List[str] | None)
roles (List[str] | None)

Return type:

TokenPair

verify_token(token: str) → Dict[str, Any][source]

Verify and decode JWT token.

Returns:: Decoded token payload or raises exception
Parameters:: token (str)
Return type:: Dict[str, Any]

refresh_access_token(refresh_token: str) → TokenPair[source]

Create new access token using refresh token.

Parameters:: refresh_token (str) – Valid refresh token
Returns:: New token pair with refreshed access token
Return type:: TokenPair

revoke_token(token: str)[source]

Revoke a token via the revocation store.

With a shared store this propagates to every worker that shares it; with the default in-memory store it is process-local (issue #1356).

Parameters:: token (str)

revoke_refresh_token(jti: str)[source]

Revoke specific refresh token.

Parameters:: jti (str)

revoke_all_user_tokens(user_id: str)[source]

Revoke all tokens for a specific user.

Parameters:: user_id (str)

cleanup_expired_tokens()[source]: Remove expired tokens from tracking.

get_public_key_jwks() → Dict[str, Any][source]

Get public key in JWKS format for external verification.

Return type:: Dict[str, Any]

get_stats() → Dict[str, Any][source]

Get authentication manager statistics.

Note: blacklisted_tokens retains the legacy field name (the config flag enable_blacklist likewise) for backward compatibility; the underlying mechanism is the revocation store. The value is the store’s count() — an int for the in-memory default, but it MAY be None for a shared backend that cannot cheaply count (per the TokenRevocationStore.count contract), meaning “unknown”.

Return type:: Dict[str, Any]

Parameters:

config (JWTConfig | None)
secret_key (str | None)
algorithm (str | None)
use_rsa (bool | None)
revocation_store (TokenRevocationStore | None)

JWT-based authentication system with token management and validation.

Usage Example:

from kailash.middleware.auth.jwt_auth import JWTAuthManager

auth_manager = JWTAuthManager(
    secret_key="your-secret-key",
    algorithm="HS256",
    access_token_expire_minutes=30
)

# Create token
token = await auth_manager.create_access_token(
    user_id="user123",
    permissions=["read", "write"]
)

# Verify token
payload = await auth_manager.verify_token(token)

Access Control Manager

Unified access control system supporting RBAC, ABAC, and hybrid strategies.

MCP Integration

MCP Server

Enhanced MCP server with caching, metrics, and configuration management.

MCP Client

Robust MCP client with connection management and error handling.

MCP Tool Node

MCP tool integration as SDK nodes for workflow usage.

MCP Resource Node

MCP resource access as SDK nodes for data integration.

Database Layer

Database Manager

Database connection and transaction management for middleware persistence.

Workflow Repository

Repository for workflow persistence and retrieval.

Database Models

Schema Generation

Node Schema Generator

Dynamic schema generation for all SDK nodes, enabling frontend node palette creation.

Dynamic Schema Registry

Caching and optimization for schema queries.

Usage Patterns

Basic Middleware Stack

from kailash.middleware import (
    AgentUIMiddleware,
    APIGateway,
    create_gateway,
    RealtimeMiddleware
)

# Create basic stack
agent_ui = AgentUIMiddleware(max_sessions=1000)
gateway = create_gateway(title="My App")
gateway.agent_ui = agent_ui

# Add real-time communication
realtime = RealtimeMiddleware(agent_ui)

# Start server
gateway.run(port=8000)

Production Configuration

import os
from kailash.middleware import create_gateway

# Environment-based configuration
gateway = create_gateway(
    title=os.getenv("APP_TITLE", "Production App"),
    cors_origins=os.getenv("CORS_ORIGINS", "").split(","),
    enable_docs=os.getenv("DEBUG", "false").lower() == "true",
    enable_auth=True,
    jwt_secret=os.getenv("JWT_SECRET"),
    max_sessions=int(os.getenv("MAX_SESSIONS", "1000")),
    session_timeout_minutes=int(os.getenv("SESSION_TIMEOUT", "60"))
)

Frontend Integration

// WebSocket connection for real-time updates
const ws = new WebSocket('ws://localhost:8000/ws?session_id=my-session');

ws.onmessage = (event) => {
    const update = JSON.parse(event.data);
    console.log('Workflow update:', update);
};

// Create and execute workflow via REST API
const response = await fetch('/api/workflows', {
    method: 'POST',
    headers: { 'Content-Type': 'application/json' },
    body: JSON.stringify({
        session_id: 'my-session',
        workflow_config: {
            nodes: [...],
            connections: [...]
        }
    })
});

Migration Guide

From Legacy API to Middleware

Old Pattern (Deprecated):

# ❌ OLD - Don't use
from kailash.api.gateway import WorkflowAPIGateway

gateway = WorkflowAPIGateway(title="App")
gateway.register_workflow("process", workflow)

New Pattern (Current):

# ✅ NEW - Use this
from kailash.middleware import create_gateway

gateway = create_gateway(title="App")
# Workflows created dynamically via API

Breaking Changes in v0.4.0

Import Paths: Change imports from kailash.api to kailash.middleware
Gateway Creation: Use create_gateway() instead of direct class instantiation
Workflow Registration: Workflows now created dynamically instead of pre-registered
Authentication: JWT authentication now integrated, not separate

Performance Considerations

Session Management

Memory Usage: Each session consumes ~1-5MB depending on workflow complexity
Cleanup: Automatic session cleanup after timeout (default 60 minutes)
Limits: Default maximum 1000 concurrent sessions (configurable)

Real-time Communication

WebSocket Connections: ~100KB memory per connection
Event Batching: Events batched for efficiency (default 100 events/batch)
Message Size: Maximum 10MB per WebSocket message

Database Performance

Connection Pooling: Default pool size 20 connections
Query Optimization: Indexed queries for workflow and execution lookups
Persistence: Optional - can run in-memory for development

Testing

Unit Testing

import pytest
from kailash.middleware import AgentUIMiddleware

@pytest.mark.asyncio
async def test_session_creation():
    agent_ui = AgentUIMiddleware()
    session_id = await agent_ui.create_session("test_user")
    assert session_id is not None

    session = await agent_ui.get_session(session_id)
    assert session.user_id == "test_user"

Integration Testing

@pytest.mark.asyncio
async def test_workflow_execution():
    agent_ui = AgentUIMiddleware()
    session_id = await agent_ui.create_session("test_user")

    workflow_config = {
        "name": "test_workflow",
        "nodes": [
            {
                "id": "test_node",
                "type": "PythonCodeNode",
                "config": {
                    "name": "test",
                    "code": "result = {'test': True}"
                }
            }
        ],
        "connections": []
    }

    workflow_id = await agent_ui.create_dynamic_workflow(
        session_id, workflow_config
    )

    execution_id = await agent_ui.execute_workflow(
        session_id, workflow_id, inputs={}
    )

    # Wait for completion and verify results
    results = await agent_ui.get_execution_results(
        session_id, execution_id
    )

    assert results["test_node"]["result"]["test"] is True

Middleware Components

Overview

Core Components

Agent-UI Middleware

API Gateway

Real-time Middleware

Event System

Event Stream

Event Types

Event Filtering

Event Classes

Authentication & Security

JWT Authentication Manager

Token revocation in multi-worker deployments

Access Control Manager

MCP Integration

MCP Server

MCP Client

MCP Tool Node

MCP Resource Node

Database Layer

Database Manager

Workflow Repository

Database Models

Schema Generation

Node Schema Generator

Dynamic Schema Registry

Usage Patterns

Basic Middleware Stack

Production Configuration

Frontend Integration

Migration Guide

From Legacy API to Middleware

Breaking Changes in v0.4.0

Performance Considerations

Session Management

Real-time Communication

Database Performance

Testing

Unit Testing

Integration Testing

Related Documentation