Middleware¶

Middleware lets you compose reusable behaviors around task execution. Use it for cross-cutting concerns like logging, error handling, or RAG context injection.

Overview¶

Middleware wraps your task handler:

Request → Middleware A → Middleware B → Handler → Middleware B → Middleware A → Response

Using Middleware¶

Pass middleware to AgentKit:

from redis_agent_kit import AgentKit, EmitterMiddleware, ResultMiddleware

kit = AgentKit(
    "redis://localhost:6379",
    agent_callable=my_handler,
    middleware=[
        ResultMiddleware(),
        EmitterMiddleware(start_message="Processing..."),
    ],
)

Middleware executes in order: ResultMiddleware wraps EmitterMiddleware wraps my_handler.

Built-in Middleware¶

ResultMiddleware¶

Automatically sets task result on success and error on exception:

from redis_agent_kit import ResultMiddleware

ResultMiddleware()  # Sets result/error and updates status

EmitterMiddleware¶

Emits progress updates at start and end:

from redis_agent_kit import EmitterMiddleware

EmitterMiddleware(
    start_message="Starting...",  # Optional
    end_message="Done",           # Optional
)

When streaming is enabled (StreamConfig(enabled=True)), these updates — along with any ctx.emitter.emit(...) / emit_token(...) calls inside your handler — are pushed to SSE clients in real time. See the Streaming guide.

RAGMiddleware¶

Searches the vector store and injects context:

from redis_agent_kit import RAGMiddleware

RAGMiddleware(
    limit=5,           # Number of results
    emit_status=True,  # Emit "Searching..." update
)

Your handler receives ctx.rag_context:

async def handler(ctx):
    print(ctx.rag_context)  # List of relevant chunks

MemoryMiddleware¶

Saves assistant responses to memory:

from redis_agent_kit import MemoryMiddleware

MemoryMiddleware(response_key="response")  # Key in result dict

SubTaskFanInMiddleware¶

Registered automatically by AgentKit when SubTaskConfig.enabled=True. Runs outside ResultMiddleware (i.e., after the terminal status is written) and:

Looks up the running task's parent via the rak:task:{child}:parent reverse-lookup.
Atomically decrements the parent's pending_children counter.
Re-enqueues the parent on its origin queue when the counter hits zero and the parent is in AWAITING_CHILDREN.

You don't construct it directly — passing subtask_config=SubTaskConfig(enabled=True) to AgentKit is enough. See the Sub-Tasks guide for the full picture.

Custom Middleware¶

Extend TaskMiddleware:

from redis_agent_kit import TaskMiddleware, TaskContext

class TimingMiddleware(TaskMiddleware):
    async def __call__(self, ctx: TaskContext, next) -> dict:
        import time
        start = time.time()
        result = await next(ctx)
        elapsed = time.time() - start
        print(f"Task {ctx.task_id} took {elapsed:.2f}s")
        return result

TaskContext¶

Middleware receives a TaskContext with:

Field	Type	Description
`task_id`	str	Task ID
`session_id`	str	Session ID
`message`	str	User message
`context`	dict	Task context
`emitter`	TaskEmitter	For progress updates
`memory`	Memory	Working and long-term memory facade
`attachments`	list	Multi-modal attachments
`rag_context`	str	Injected by RAGMiddleware
`parent_id`	str \| None	Set on child tasks (sub-tasking)
`child_ids`	list[str]	Children spawned so far (sub-tasking)
`depth`	int	Nesting depth — 0 for root (sub-tasking)

When sub-tasking is enabled, TaskContext also exposes spawn, spawn_and_wait, gather, and child_result. See the Sub-Tasks guide.

Pipeline Middleware¶

Pipelines have separate middleware:

from redis_agent_kit import PipelineLoggingMiddleware, PipelineTimingMiddleware
from redis_agent_kit.pipelines import PipelineOrchestrator

orchestrator = PipelineOrchestrator(
    config=config,
    base_path=path,
    middleware=[
        PipelineLoggingMiddleware(),
        PipelineTimingMiddleware(),
    ],
)

Built-in Pipeline Middleware¶

Middleware	Description
`PipelineLoggingMiddleware`	Logs stage start/end
`PipelineTimingMiddleware`	Tracks timing in metadata
`PipelineValidationMiddleware`	Validates config

Custom Pipeline Middleware¶

from redis_agent_kit import PipelineMiddleware, PipelineContext

class AuditMiddleware(PipelineMiddleware):
    async def __call__(self, ctx: PipelineContext, next) -> dict:
        print(f"Starting {ctx.stage}")
        result = await next(ctx)
        print(f"Finished {ctx.stage}: {result}")
        return result