Tasks & Middleware

These types are what an agent callable sees at runtime: the TaskContext passed to your function, the emitter used for progress updates, and the middleware that wraps execution.

TaskContext

TaskContext dataclass

TaskContext(task_id: str, session_id: str | None, message: str, context: dict[str, Any], kit: AgentKit, emitter: TaskEmitter, memory: Memory, attachments: list[Any] = list(), rag_context: str = '', metadata: dict[str, Any] = dict(), parent_id: str | None = None, child_ids: list[str] = list(), depth: int = 0, _resumed: bool = False, _missing_spawn_key_warned: bool = False)

Context passed through the middleware chain to the task handler.

Contains all the information needed to process a task, plus any data injected by middleware (like rag_context).

Attributes:

Name	Type	Description
task_id	str	The unique task identifier
session_id	str | None	The memory session identifier (optional)
message	str	The user's message/query
attachments	list[Any]	List of attachments (images, audio, files) for multi-modal input
context	dict[str, Any]	Additional context data (user_id, instance_id, etc.)
kit	AgentKit	The AgentKit instance for accessing managers
emitter	TaskEmitter	TaskEmitter for progress updates
memory	Memory	Memory instance for working/long-term memory
rag_context	str	RAG context injected by RAGMiddleware (optional)
metadata	dict[str, Any]	Additional data injected by custom middleware
parent_id	str | None	Parent task ID if this handler is running a child task
child_ids	list[str]	IDs of children this task has spawned (refreshed each invocation)
depth	int	Nesting depth (0 for root tasks)

spawn async

spawn(message: str, *, spawn_key: str | None = None, context: dict[str, Any] | None = None, isolate_memory: bool | None = None) -> TaskState

Spawn a child task without waiting for it.

Parameters:

Name	Type	Description	Default
message	str	Message/query for the child task.	required
spawn_key	str | None	Optional stable label tying the child to a parent handler step. See docs/proposals/subtasking-spec.md §10 for guidance.	None
context	dict[str, Any] | None	Optional context dict to pass to the child handler.	None
isolate_memory	bool | None	If True, child gets a fresh session_id (linked to parent via parent_id). If False, child shares parent's session. If None, uses the kit-wide SubTaskConfig default.	None

Returns:

Type	Description
TaskState	The child TaskState (status QUEUED).

Source code in redis_agent_kit/middleware.py

async def spawn(
    self,
    message: str,
    *,
    spawn_key: str | None = None,
    context: dict[str, Any] | None = None,
    isolate_memory: bool | None = None,
) -> TaskState:
    """Spawn a child task without waiting for it.

    Args:
        message: Message/query for the child task.
        spawn_key: Optional stable label tying the child to a parent
            handler step. See docs/proposals/subtasking-spec.md §10 for guidance.
        context: Optional context dict to pass to the child handler.
        isolate_memory: If True, child gets a fresh session_id (linked to
            parent via parent_id). If False, child shares parent's session.
            If None, uses the kit-wide SubTaskConfig default.

    Returns:
        The child TaskState (status QUEUED).
    """
    return await self.kit._spawn_child(
        parent_ctx=self,
        message=message,
        spawn_key=spawn_key,
        context=context,
        isolate_memory=isolate_memory,
    )

spawn_and_wait async

spawn_and_wait(message: str, *, spawn_key: str | None = None, context: dict[str, Any] | None = None, isolate_memory: bool | None = None, on_failure: Literal['propagate', 'collect', 'ignore'] | None = None) -> Any

Spawn a child and pause the parent until it completes.

First invocation: spawns the child, parks the parent in AWAITING_CHILDREN, and aborts the handler (the framework swallows the control-flow signal; the parent will be re-enqueued by the fan-in hook).

Resume invocation: if a child has already been spawned for this spawn_key and is terminal, returns its result; if it failed, applies the configured on_failure mode. If the prior child is still running, parks again.

Parameters:

Name	Type	Description	Default
message	str	Message/query for the child task.	required
spawn_key	str | None	Stable label tying the child to this step. Strongly recommended; without it, every re-entry spawns a fresh child.	None
context	dict[str, Any] | None	Optional context dict.	None
isolate_memory	bool | None	See spawn().	None
on_failure	Literal['propagate', 'collect', 'ignore'] | None	How to handle a failed child. Defaults to the kit's SubTaskConfig.default_on_failure.	None

Returns:

Type	Description
Any	The child's result dict on success. With on_failure="ignore",
Any	returns None for failed children. With on_failure="collect"
Any	on a single-child wait, returns {"error": ...} instead of raising.

Source code in redis_agent_kit/middleware.py

async def spawn_and_wait(
    self,
    message: str,
    *,
    spawn_key: str | None = None,
    context: dict[str, Any] | None = None,
    isolate_memory: bool | None = None,
    on_failure: Literal["propagate", "collect", "ignore"] | None = None,
) -> Any:
    """Spawn a child and pause the parent until it completes.

    First invocation: spawns the child, parks the parent in
    AWAITING_CHILDREN, and aborts the handler (the framework swallows
    the control-flow signal; the parent will be re-enqueued by the
    fan-in hook).

    Resume invocation: if a child has already been spawned for this
    ``spawn_key`` and is terminal, returns its result; if it failed,
    applies the configured ``on_failure`` mode. If the prior child is
    still running, parks again.

    Args:
        message: Message/query for the child task.
        spawn_key: Stable label tying the child to this step. Strongly
            recommended; without it, every re-entry spawns a fresh child.
        context: Optional context dict.
        isolate_memory: See spawn().
        on_failure: How to handle a failed child. Defaults to the kit's
            SubTaskConfig.default_on_failure.

    Returns:
        The child's result dict on success. With ``on_failure="ignore"``,
        returns ``None`` for failed children. With ``on_failure="collect"``
        on a single-child wait, returns ``{"error": ...}`` instead of raising.
    """
    # If we already spawned a child with this spawn_key on a prior
    # invocation, look up its current state and return/raise/re-pause.
    existing = await self._existing_child_for_spawn_key(spawn_key)
    if existing is not None:
        outcome = await self._resolve_child_outcome(existing, spawn_key, on_failure)
        if outcome is _STILL_PENDING:
            raise _SpawnAwaitSignal([_SpawnAwait(existing.task_id, spawn_key)])
        return outcome

    # First time: warn if re-entrant + missing spawn_key, then spawn.
    if spawn_key is None and self._resumed and not self._missing_spawn_key_warned:
        warnings.warn(
            "RAK: spawn_and_wait called without spawn_key in a re-entrant handler. "
            "This will spawn a fresh child on every resume. If that's intentional, "
            "pass spawn_key=None explicitly. See docs/proposals/subtasking-spec.md §10.",
            UserWarning,
            stacklevel=2,
        )
        self._missing_spawn_key_warned = True

    child = await self.spawn(
        message=message,
        spawn_key=spawn_key,
        context=context,
        isolate_memory=isolate_memory,
    )
    raise _SpawnAwaitSignal([_SpawnAwait(child.task_id, spawn_key)])

gather async

gather(specs: list[SpawnSpec], on_failure: Literal['propagate', 'collect', 'ignore'] | None = None) -> list[Any]

Spawn multiple children concurrently and pause until all complete.

Uses positional indexing (_gather:<i>) as a fallback spawn_key when a spec does not provide one, so resumption stays idempotent.

Parameters:

Name	Type	Description	Default
specs	list[SpawnSpec]	One SpawnSpec per child to spawn.	required
on_failure	Literal['propagate', 'collect', 'ignore'] | None	Failure-propagation mode applied to each child.	None

Returns:

Type	Description
list[Any]	Results list, in input order. Failed children are represented
list[Any]	per the on_failure mode (None for ignore, error dict for collect).

Source code in redis_agent_kit/middleware.py

async def gather(
    self,
    specs: list[SpawnSpec],
    on_failure: Literal["propagate", "collect", "ignore"] | None = None,
) -> list[Any]:
    """Spawn multiple children concurrently and pause until all complete.

    Uses positional indexing (``_gather:<i>``) as a fallback spawn_key
    when a spec does not provide one, so resumption stays idempotent.

    Args:
        specs: One SpawnSpec per child to spawn.
        on_failure: Failure-propagation mode applied to each child.

    Returns:
        Results list, in input order. Failed children are represented
        per the ``on_failure`` mode (None for ignore, error dict for collect).
    """
    results: list[Any] = []
    pending_awaits: list[_SpawnAwait] = []
    any_new_spawn = False

    for index, spec in enumerate(specs):
        key = spec.spawn_key or f"_gather:{index}"
        existing = await self._existing_child_for_spawn_key(key)
        if existing is not None:
            outcome = await self._resolve_child_outcome(existing, key, on_failure)
            if outcome is _STILL_PENDING:
                pending_awaits.append(_SpawnAwait(existing.task_id, key))
                results.append(None)
            else:
                results.append(outcome)
            continue

        # Spawn fresh
        child = await self.spawn(
            message=spec.message,
            spawn_key=key,
            context=spec.context,
            isolate_memory=spec.isolate_memory,
        )
        pending_awaits.append(_SpawnAwait(child.task_id, key))
        results.append(None)
        any_new_spawn = True

    if pending_awaits:
        # Park until all spawned children finish. ``any_new_spawn`` is
        # informational — fan-in cares about the live pending counter.
        del any_new_spawn  # quiet unused-var
        raise _SpawnAwaitSignal(pending_awaits)

    return results

child_result async

child_result(spawn_key: str) -> Any | None

Read a completed child's result by its spawn_key.

Returns None if no child with that spawn_key exists yet or the child is not in a terminal state.

Source code in redis_agent_kit/middleware.py

async def child_result(self, spawn_key: str) -> Any | None:
    """Read a completed child's result by its spawn_key.

    Returns None if no child with that spawn_key exists yet or the child
    is not in a terminal state.
    """
    existing = await self._existing_child_for_spawn_key(spawn_key)
    if existing is None:
        return None
    if existing.status != TaskStatus.DONE:
        return None
    return existing.result

TaskEmitter

TaskEmitter

TaskEmitter(task_manager: TaskManager, task_id: str)

Emits progress updates during task execution.

This is the primary interface for agent code to send real-time updates about what it's doing. Updates are stored on the task and can be polled by clients.

Example

async def my_agent(query: str, emitter: TaskEmitter): await emitter.emit("Searching knowledge base...") results = await search(query) await emitter.emit(f"Found {len(results)} results", update_type="info") return {"answer": synthesize(results)}

Initialize TaskEmitter.

Parameters:

Name	Type	Description	Default
task_manager	TaskManager	The TaskManager for storing updates	required
task_id	str	The task ID to emit updates for	required

Source code in redis_agent_kit/emitter.py

def __init__(self, task_manager: TaskManager, task_id: str):
    """Initialize TaskEmitter.

    Args:
        task_manager: The TaskManager for storing updates
        task_id: The task ID to emit updates for
    """
    self._task_manager = task_manager
    self._task_id = task_id

task_id property

task_id: str

Get the task ID.

task_manager property

task_manager: TaskManager

Get the task manager.

emit async

emit(message: str, update_type: str = 'info', metadata: dict[str, Any] | None = None) -> None

Emit a progress update.

Parameters:

Name	Type	Description	Default
message	str	The update message	required
update_type	str	Type of update (info, tool_call, reflection, error)	'info'
metadata	dict[str, Any] | None	Optional additional metadata	None

Source code in redis_agent_kit/emitter.py

async def emit(
    self,
    message: str,
    update_type: str = "info",
    metadata: dict[str, Any] | None = None,
) -> None:
    """Emit a progress update.

    Args:
        message: The update message
        update_type: Type of update (info, tool_call, reflection, error)
        metadata: Optional additional metadata
    """
    await self._task_manager.add_update(
        self._task_id,
        message,
        update_type=update_type,
        metadata=metadata,
    )

emit_nowait

emit_nowait(message: str, update_type: str = 'info', metadata: dict[str, Any] | None = None) -> None

Emit a progress update without waiting (fire-and-forget).

This schedules the update to be stored without blocking. Useful when you don't need to await the storage operation.

Parameters:

Name	Type	Description	Default
message	str	The update message	required
update_type	str	Type of update (info, tool_call, reflection, error)	'info'
metadata	dict[str, Any] | None	Optional additional metadata	None

Source code in redis_agent_kit/emitter.py

def emit_nowait(
    self,
    message: str,
    update_type: str = "info",
    metadata: dict[str, Any] | None = None,
) -> None:
    """Emit a progress update without waiting (fire-and-forget).

    This schedules the update to be stored without blocking.
    Useful when you don't need to await the storage operation.

    Args:
        message: The update message
        update_type: Type of update (info, tool_call, reflection, error)
        metadata: Optional additional metadata
    """
    # Schedule the async operation without waiting
    asyncio.create_task(self.emit(message, update_type=update_type, metadata=metadata))

emit_token async

emit_token(token: str, metadata: dict[str, Any] | None = None) -> None

Emit a single token via Pub/Sub without persisting to task state.

This is a lightweight, publish-only path for token-by-token streaming. Unlike emit(), it does NOT read or write the task object in Redis — it only publishes to the Pub/Sub channel. Requires enable_streaming=True on the TaskManager.

Parameters:

Name	Type	Description	Default
token	str	The token string to publish	required
metadata	dict[str, Any] | None	Optional metadata dict	None

Source code in redis_agent_kit/emitter.py

async def emit_token(
    self,
    token: str,
    metadata: dict[str, Any] | None = None,
) -> None:
    """Emit a single token via Pub/Sub without persisting to task state.

    This is a lightweight, publish-only path for token-by-token streaming.
    Unlike emit(), it does NOT read or write the task object in Redis —
    it only publishes to the Pub/Sub channel. Requires
    ``enable_streaming=True`` on the TaskManager.

    Args:
        token: The token string to publish
        metadata: Optional metadata dict
    """
    await self._task_manager.publish(
        self._task_id,
        token,
        event_type="token",
        metadata=metadata,
    )

emit_token_nowait

emit_token_nowait(token: str, metadata: dict[str, Any] | None = None) -> None

Emit a single token without waiting (fire-and-forget).

Non-blocking variant of emit_token(). Schedules the publish without awaiting it.

Parameters:

Name	Type	Description	Default
token	str	The token string to publish	required
metadata	dict[str, Any] | None	Optional metadata dict	None

Source code in redis_agent_kit/emitter.py

def emit_token_nowait(
    self,
    token: str,
    metadata: dict[str, Any] | None = None,
) -> None:
    """Emit a single token without waiting (fire-and-forget).

    Non-blocking variant of emit_token(). Schedules the publish
    without awaiting it.

    Args:
        token: The token string to publish
        metadata: Optional metadata dict
    """
    asyncio.create_task(self.emit_token(token, metadata=metadata))

TaskMiddleware

TaskMiddleware

Bases: ABC

Base class for task handler middleware.

Middleware wraps task handlers to add cross-cutting functionality. Each middleware receives a TaskContext and a 'next' function to call the next middleware in the chain (or the final handler).

Example

class LoggingMiddleware(TaskMiddleware): async def call(self, ctx: TaskContext, next: NextHandler) -> dict: print(f"Starting task {ctx.task_id}") result = await next(ctx) print(f"Completed task {ctx.task_id}") return result

MiddlewareChain

MiddlewareChain

MiddlewareChain(handler: Callable[[TaskContext], Coroutine[Any, Any, dict[str, Any]]], middleware: list[TaskMiddleware] | None = None)

Chains multiple middleware together with a final handler.

The chain is built from the inside out: - The innermost function is the actual task handler - Each middleware wraps the previous, forming a chain - When called, execution flows: middleware1 -> middleware2 -> ... -> handler

Initialize the middleware chain.

Parameters:

Name	Type	Description	Default
handler	Callable[[TaskContext], Coroutine[Any, Any, dict[str, Any]]]	The final task handler function	required
middleware	list[TaskMiddleware] | None	List of middleware to apply (in order)	None

Source code in redis_agent_kit/middleware.py

def __init__(
    self,
    handler: Callable[[TaskContext], Coroutine[Any, Any, dict[str, Any]]],
    middleware: list[TaskMiddleware] | None = None,
):
    """Initialize the middleware chain.

    Args:
        handler: The final task handler function
        middleware: List of middleware to apply (in order)
    """
    self._handler = handler
    self._middleware = middleware or []

EmitterMiddleware

EmitterMiddleware

EmitterMiddleware(start_message: str = 'Starting...', end_message: str | None = None)

Bases: TaskMiddleware

Middleware that emits progress updates at start and end.

Automatically emits "Starting..." when the task begins and handles the status update pattern that most handlers need.

Parameters:

Name	Type	Description	Default
start_message	str	Message to emit when starting (default: "Starting...")	'Starting...'
end_message	str | None	Message to emit on success (default: None, no message)	None

Source code in redis_agent_kit/middleware.py

def __init__(
    self,
    start_message: str = "Starting...",
    end_message: str | None = None,
):
    self.start_message = start_message
    self.end_message = end_message

MemoryMiddleware

MemoryMiddleware

MemoryMiddleware(response_key: str = 'response', role: str = 'assistant')

Bases: TaskMiddleware

Middleware that saves the response to working memory.

After the handler completes, if the result contains a 'response' key, it's saved as an assistant message in memory.

Parameters:

Name	Type	Description	Default
response_key	str	Key in result dict containing the response (default: "response")	'response'
role	str	Message role to use (default: "assistant")	'assistant'

Source code in redis_agent_kit/middleware.py

def __init__(self, response_key: str = "response", role: str = "assistant"):
    self.response_key = response_key
    self.role = role

RAGMiddleware

RAGMiddleware

RAGMiddleware(limit: int = 3, separator: str = '\n\n', emit_status: bool = True)

Bases: TaskMiddleware

Middleware that injects RAG context from the vector store.

Searches the vector store for relevant documents based on the user's message and injects them into ctx.rag_context.

Parameters:

Name	Type	Description	Default
limit	int	Maximum number of results to retrieve (default: 3)	3
separator	str	String to join multiple results (default: "\n\n")	'\n\n'
emit_status	bool	Whether to emit a status update (default: True)	True

Source code in redis_agent_kit/middleware.py

def __init__(
    self,
    limit: int = 3,
    separator: str = "\n\n",
    emit_status: bool = True,
):
    self.limit = limit
    self.separator = separator
    self.emit_status = emit_status

ResultMiddleware

ResultMiddleware

Bases: TaskMiddleware

Middleware that sets the task result/error status.

Automatically calls set_result on success or set_error on exception. This is the outermost middleware that handles the task lifecycle.

Note: If the task is in AWAITING_INPUT or AWAITING_CHILDREN status (paused), the result is stored but status is not changed to DONE.

Sub-tasking

Spawn and coordinate child tasks. See the Sub-Tasks guide for the full walkthrough.

SubTaskConfig

SubTaskConfig

Bases: BaseSettings

Configuration for the sub-tasking subsystem.

Sub-tasking is opt-in: when disabled, ctx.spawn* methods raise a clear error and no fan-in middleware is registered. When enabled, handlers can spawn child tasks that run on a separate Docket queue and the parent automatically pauses/resumes around them.

Environment variables: RAK_SUBTASK__* (e.g. RAK_SUBTASK__ENABLED=true).

enabled class-attribute instance-attribute

enabled: bool = False

Master switch. When False, ctx.spawn* raises NotImplementedError.

max_depth class-attribute instance-attribute

max_depth: int = 5

Maximum nesting depth (root is depth 0). Guards against runaway recursion.

max_children_per_task class-attribute instance-attribute

max_children_per_task: int = 32

Maximum number of children any single task may spawn.

default_on_failure class-attribute instance-attribute

default_on_failure: Literal['propagate', 'collect', 'ignore'] = 'propagate'

Default failure-propagation mode for child tasks awaited by the parent.

cascade_cancel class-attribute instance-attribute

cascade_cancel: bool = True

When True, cancelling a parent cascades cancellation to all descendants.

child_queue_name class-attribute instance-attribute

child_queue_name: str | None = None

Docket queue name for children. Defaults to <parent_queue>_children.

isolate_memory class-attribute instance-attribute

isolate_memory: bool = False

When True, children get their own session_id by default (linked to parent via parent_id, not session membership). Per-spawn override always wins.

SpawnSpec

SpawnSpec dataclass

SpawnSpec(message: str, spawn_key: str | None = None, context: dict[str, Any] | None = None, isolate_memory: bool | None = None)

Specification for one child task to be spawned via ctx.gather.

SubTaskFanInMiddleware

SubTaskFanInMiddleware

Bases: TaskMiddleware

Middleware that runs the fan-in hook on terminal child transitions.

Registered automatically when SubTaskConfig.enabled = True. After the inner chain finishes (success or failure), this looks up the running task's parent (via the rak:task:{child}:parent reverse-lookup), atomic DECRs the parent's pending counter, and re-enqueues the parent if it has reached zero pending children.

It MUST run outside of (after) ResultMiddleware — the parent's pending counter should only decrement once the child has been written to its terminal status.