Core

The core module wires an agent callable to Redis-backed durable tasks and background workers.

AgentKit

AgentKit

AgentKit(redis_url: str | None = None, *, redis_client: Redis | None = None, prefix: str | None = None, agent_callable: AgentCallable | ContextAgentCallable | None = None, middleware: list[TaskMiddleware] | None = None, queue_name: str | None = None, settings: Settings | None = None, auto_result_middleware: bool = True, enable_sse: bool = False, enable_streaming: bool = False, stream_config: StreamConfig | None = None, subtask_config: SubTaskConfig | None = None)

Convenience class that bundles TaskManager, Memory, and helpers.

This is the main entry point for using Redis Agent Kit in your application. Docket integration is internal - users only interact with tasks and memory.

Example

async def my_agent(ctx: TaskContext) -> dict: # ctx.memory is always available await ctx.memory.add_message("assistant", "Processing...") return {"response": f"Processed: {ctx.message}"}

kit = AgentKit("redis://localhost:6379", agent_callable=my_agent)

Example (with middleware): from redis_agent_kit.middleware import RAGMiddleware, MemoryMiddleware

async def my_agent(ctx: TaskContext) -> dict:
    # ctx.rag_context is injected by RAGMiddleware
    return {"response": f"Based on context: {ctx.rag_context}"}

kit = AgentKit(
    "redis://localhost:6379",
    agent_callable=my_agent,
    middleware=[RAGMiddleware(), MemoryMiddleware()],
)

Initialize AgentKit.

Parameters:

Name	Type	Description	Default
redis_url	str | None	Redis URL (e.g., "redis://localhost:6379"). Preferred.	None
redis_client	Redis | None	Optional pre-created async Redis client. If provided, redis_url is still needed for background workers.	None
prefix	str | None	Optional custom key prefix (default: 'rak')	None
agent_callable	AgentCallable | ContextAgentCallable | None	Your agent function. Can be either: - (task_id, session_id, message, context) -> dict - (ctx: TaskContext) -> dict (use with middleware)	None
middleware	list[TaskMiddleware] | None	List of TaskMiddleware to apply (in order)	None
queue_name	str | None	Background worker queue name (default from settings)	None
settings	Settings | None	Optional Settings instance (uses global settings if not provided)	None
auto_result_middleware	bool	Automatically include ResultMiddleware (default: True). Set to False if you want full control over middleware.	True
enable_sse	bool	Enable Pub/Sub publishing for SSE streaming (default: False). Deprecated – use stream_config instead.	False
enable_streaming	bool	Enable lightweight Pub/Sub publishing for token-by-token streaming (default: False). Deprecated – use stream_config instead.	False
stream_config	StreamConfig | None	Structured streaming configuration. When provided, enable_sse and enable_streaming are ignored.	None

Example

Simple - just URL

kit = AgentKit("redis://localhost:6379", agent_callable=my_agent)

With existing client

client = redis.from_url("redis://localhost:6379") kit = AgentKit("redis://localhost:6379", redis_client=client, ...)

With StreamConfig

kit = AgentKit( "redis://localhost:6379", stream_config=StreamConfig(enabled=True, channels={ChannelScope.TASK, ChannelScope.SESSION}), )

Source code in redis_agent_kit/core.py

def __init__(
    self,
    redis_url: str | None = None,
    *,
    redis_client: redis_lib.Redis | None = None,
    prefix: str | None = None,
    agent_callable: AgentCallable | ContextAgentCallable | None = None,
    middleware: list[TaskMiddleware] | None = None,
    queue_name: str | None = None,
    settings: Settings | None = None,
    auto_result_middleware: bool = True,
    enable_sse: bool = False,
    enable_streaming: bool = False,
    stream_config: StreamConfig | None = None,
    subtask_config: SubTaskConfig | None = None,
):
    """Initialize AgentKit.

    Args:
        redis_url: Redis URL (e.g., "redis://localhost:6379"). Preferred.
        redis_client: Optional pre-created async Redis client. If provided,
            redis_url is still needed for background workers.
        prefix: Optional custom key prefix (default: 'rak')
        agent_callable: Your agent function. Can be either:
            - (task_id, session_id, message, context) -> dict
            - (ctx: TaskContext) -> dict (use with middleware)
        middleware: List of TaskMiddleware to apply (in order)
        queue_name: Background worker queue name (default from settings)
        settings: Optional Settings instance (uses global settings if not provided)
        auto_result_middleware: Automatically include ResultMiddleware (default: True).
            Set to False if you want full control over middleware.
        enable_sse: Enable Pub/Sub publishing for SSE streaming (default: False).
            Deprecated – use ``stream_config`` instead.
        enable_streaming: Enable lightweight Pub/Sub publishing for
            token-by-token streaming (default: False).
            Deprecated – use ``stream_config`` instead.
        stream_config: Structured streaming configuration. When provided,
            ``enable_sse`` and ``enable_streaming`` are ignored.

    Example:
        # Simple - just URL
        kit = AgentKit("redis://localhost:6379", agent_callable=my_agent)

        # With existing client
        client = redis.from_url("redis://localhost:6379")
        kit = AgentKit("redis://localhost:6379", redis_client=client, ...)

        # With StreamConfig
        kit = AgentKit(
            "redis://localhost:6379",
            stream_config=StreamConfig(enabled=True, channels={ChannelScope.TASK, ChannelScope.SESSION}),
        )
    """
    from redis_agent_kit.middleware import ResultMiddleware

    self._settings = settings or get_settings()

    # Resolve Redis URL
    self._redis_url = redis_url or self._settings.redis_url
    if not self._redis_url:
        raise ValueError(
            "Redis URL required. Pass redis_url or set RAK_REDIS_URL environment variable."
        )

    # Create or use provided client
    if redis_client is not None:
        self._redis = redis_client
    else:
        self._redis = redis_lib.from_url(self._redis_url)  # type: ignore[no-untyped-call]

    self._prefix = prefix or self._settings.namespace
    self._agent_callable = agent_callable
    self._queue_name = queue_name or self._settings.task.queue_name

    # Resolve StreamConfig
    if stream_config is not None:
        self._stream_config = stream_config
    else:
        self._stream_config = StreamConfig.from_booleans(enable_sse, enable_streaming)

    # Legacy accessors for backward compat
    self._enable_sse = self._stream_config.enabled
    self._enable_streaming = self._stream_config.enabled

    self._task_manager = TaskManager(
        self._redis,
        prefix=self._prefix,
        stream_config=self._stream_config,
    )

    # Resolve SubTaskConfig (explicit arg wins over settings)
    self._subtask_config = subtask_config or self._settings.subtask
    # Resolve the children queue name now so spawn dispatch can read it.
    self._child_queue_name = (
        self._subtask_config.child_queue_name
        if self._subtask_config.child_queue_name
        else f"{self._queue_name}_children"
    )

    # Build middleware list
    user_middleware = middleware or []

    # Auto-include ResultMiddleware unless user opts out or already has it
    if auto_result_middleware:
        has_result_middleware = any(isinstance(m, ResultMiddleware) for m in user_middleware)
        if not has_result_middleware:
            # ResultMiddleware should be outermost (first in list)
            self._middleware = [ResultMiddleware(), *user_middleware]
        else:
            self._middleware = user_middleware
    else:
        self._middleware = user_middleware

    # When subtasking is enabled, prepend SubTaskFanInMiddleware so it runs
    # *outside* ResultMiddleware (i.e., after the terminal status has been
    # written). It's a no-op for tasks without a parent.
    if self._subtask_config.enabled:
        has_fanin = any(isinstance(m, SubTaskFanInMiddleware) for m in self._middleware)
        if not has_fanin:
            self._middleware = [SubTaskFanInMiddleware(), *self._middleware]

    # Initialize memory (always available, may be no-op if disabled)
    if self._settings.memory.enabled:
        _require_ams()
        self._memory = Memory(self._redis, namespace=self._settings.namespace)
    else:
        self._memory = NoOpMemory()

task_manager property

task_manager: TaskManager

Get the TaskManager.

memory property

memory: Memory

Get the Memory instance.

queue_name property

queue_name: str

Get the background worker queue name.

child_queue_name property

child_queue_name: str

Docket queue name used for spawned child tasks.

subtask_config property

subtask_config: SubTaskConfig

Active sub-tasking configuration.

redis_url property

redis_url: str

Get the Redis URL for background workers.

worker_task property

worker_task: Callable[[str, str, str, dict[str, Any]], Coroutine[Any, Any, dict[str, Any]]]

Get the worker task handler for use with Docket.

This is the public API for getting a handler suitable for background workers. Use this instead of _create_wrapped_handler().

Example

kit = AgentKit("redis://localhost:6379", agent_callable=my_agent)

For Docket workers:

tasks = [kit.worker_task]

Returns:

Type	Description
Callable[[str, str, str, dict[str, Any]], Coroutine[Any, Any, dict[str, Any]]]	Wrapped async function suitable for Docket workers

Raises:

Type	Description
ValueError	If no agent_callable is configured

create_task async

create_task(message: str, context: dict[str, Any] | None = None, session_id: str | None = None, user_id: str | None = None) -> tuple[TaskState, str]

Create a new task with a memory session.

Parameters:

Name	Type	Description	Default
message	str	The initial user message	required
context	dict[str, Any] | None	Optional context data	None
session_id	str | None	Optional existing session ID	None
user_id	str | None	Optional user ID for memory isolation	None

Returns:

Type	Description
tuple[TaskState, str]	Tuple of (TaskState, session_id)

Source code in redis_agent_kit/core.py

async def create_task(
    self,
    message: str,
    context: dict[str, Any] | None = None,
    session_id: str | None = None,
    user_id: str | None = None,
) -> tuple[TaskState, str]:
    """Create a new task with a memory session.

    Args:
        message: The initial user message
        context: Optional context data
        session_id: Optional existing session ID
        user_id: Optional user ID for memory isolation

    Returns:
        Tuple of (TaskState, session_id)
    """
    return await create_task_with_session(
        task_manager=self._task_manager,
        memory=self._memory,
        message=message,
        context=context,
        session_id=session_id,
        user_id=user_id,
    )

create_and_submit_task async

create_and_submit_task(message: str, context: dict[str, Any] | None = None, session_id: str | None = None, caller: Caller | None = None, user_id: str | None = None, attachments: list[Any] | None = None) -> dict[str, Any]

Create a task and submit for background processing via Docket.

This is the primary method for creating tasks that will be processed asynchronously. Docket handles the background execution - users only see tasks and memory.

Parameters:

Name	Type	Description	Default
message	str	The user message/query	required
context	dict[str, Any] | None	Optional context data (user_id, instance_id, etc.)	None
session_id	str | None	Optional existing session ID	None
caller	Caller | None	Optional caller identity (who is creating this task)	None
user_id	str | None	Optional user ID for memory isolation	None
attachments	list[Any] | None	Optional list of Attachment objects for multi-modal input	None

Returns:

Type	Description
dict[str, Any]	Dict with task_id, session_id, and status

Raises:

Type	Description
ValueError	If no agent_callable is configured

Source code in redis_agent_kit/core.py

async def create_and_submit_task(
    self,
    message: str,
    context: dict[str, Any] | None = None,
    session_id: str | None = None,
    caller: Caller | None = None,
    user_id: str | None = None,
    attachments: list[Any] | None = None,
) -> dict[str, Any]:
    """Create a task and submit for background processing via Docket.

    This is the primary method for creating tasks that will be processed
    asynchronously. Docket handles the background execution - users only
    see tasks and memory.

    Args:
        message: The user message/query
        context: Optional context data (user_id, instance_id, etc.)
        session_id: Optional existing session ID
        caller: Optional caller identity (who is creating this task)
        user_id: Optional user ID for memory isolation
        attachments: Optional list of Attachment objects for multi-modal input

    Returns:
        Dict with task_id, session_id, and status

    Raises:
        ValueError: If no agent_callable is configured
    """
    if self._agent_callable is None:
        raise ValueError(
            "No agent_callable configured. Pass agent_callable to AgentKit() "
            "to use create_and_submit_task()."
        )

    # Bind memory to session
    bound_memory = self._memory.with_session(session_id, user_id)

    # Add message to memory (creates session if needed)
    await bound_memory.add_message(role="user", content=message)

    # Create task
    task = await self._task_manager.create_task(
        session_id=bound_memory.session_id,
        message=message,
        context=context,
        caller=caller,
    )

    # Create wrapped handler with middleware
    handler = self._create_wrapped_handler()

    # Prepare context with attachments
    task_context = context.copy() if context else {}
    if attachments:
        # Serialize attachments for Docket (they're Pydantic models)
        task_context["_attachments"] = [
            a.model_dump() if hasattr(a, "model_dump") else a for a in attachments
        ]

    # Submit to background worker queue for processing
    async with Docket(url=self._redis_url, name=self._queue_name) as docket:
        # Use task_id as the Docket key (enables cancellation by task_id)
        queued_func: Any = docket.add(handler, key=task.task_id)
        await queued_func(
            task_id=task.task_id,
            session_id=bound_memory.session_id,
            message=message,
            context=task_context,
        )

    return {
        "task_id": task.task_id,
        "session_id": bound_memory.session_id,
        "status": "queued",
        "message": "Task created and queued for processing",
    }

get_emitter

get_emitter(task_id: str) -> TaskEmitter

Get a TaskEmitter for a task.

Parameters:

Name	Type	Description	Default
task_id	str	The task ID	required

Returns:

Type	Description
TaskEmitter	TaskEmitter for emitting progress updates

Source code in redis_agent_kit/core.py

def get_emitter(self, task_id: str) -> TaskEmitter:
    """Get a TaskEmitter for a task.

    Args:
        task_id: The task ID

    Returns:
        TaskEmitter for emitting progress updates
    """
    return TaskEmitter(self._task_manager, task_id)

cancel_task async

cancel_task(task_id: str, *, cascade: bool | None = None) -> int

Cancel a task, optionally cascading to descendants.

Parameters:

Name	Type	Description	Default
task_id	str	The task ID to cancel.	required
cascade	bool | None	When True, cancels all descendant tasks. Defaults to the kit's SubTaskConfig.cascade_cancel when subtasking is enabled, else False.	None

Returns:

Type	Description
int	Number of tasks transitioned to CANCELLED.

Source code in redis_agent_kit/core.py

async def cancel_task(self, task_id: str, *, cascade: bool | None = None) -> int:
    """Cancel a task, optionally cascading to descendants.

    Args:
        task_id: The task ID to cancel.
        cascade: When True, cancels all descendant tasks. Defaults to the
            kit's ``SubTaskConfig.cascade_cancel`` when subtasking is
            enabled, else False.

    Returns:
        Number of tasks transitioned to CANCELLED.
    """
    if cascade is None:
        cascade = self._subtask_config.enabled and self._subtask_config.cascade_cancel
    return await self._task_manager.cancel_task(task_id, cascade=cascade)

submit_input async

submit_input(task_id: str, response: dict[str, Any]) -> bool

Submit user input to a task that is awaiting input.

This stores the user's response and re-queues the task for processing. The task handler will receive the input in the task's input_response field.

Parameters:

Name	Type	Description	Default
task_id	str	The task ID	required
response	dict[str, Any]	Dictionary mapping field names to values. For simple prompts without fields, use {"response": "..."}	required

Returns:

Type	Description
bool	True if input was submitted successfully, False if task not found
bool	or not in AWAITING_INPUT status.

Example

Simple text response

await kit.submit_input(task_id, {"response": "Yes, proceed"})

Structured response matching schema

await kit.submit_input(task_id, { "database": "Redis", "confirm": True, })

Source code in redis_agent_kit/core.py

async def submit_input(self, task_id: str, response: dict[str, Any]) -> bool:
    """Submit user input to a task that is awaiting input.

    This stores the user's response and re-queues the task for processing.
    The task handler will receive the input in the task's input_response field.

    Args:
        task_id: The task ID
        response: Dictionary mapping field names to values.
                  For simple prompts without fields, use {"response": "..."}

    Returns:
        True if input was submitted successfully, False if task not found
        or not in AWAITING_INPUT status.

    Example:
        # Simple text response
        await kit.submit_input(task_id, {"response": "Yes, proceed"})

        # Structured response matching schema
        await kit.submit_input(task_id, {
            "database": "Redis",
            "confirm": True,
        })
    """
    # Update task status and store response
    success = await self._task_manager.submit_input(task_id, response)
    if not success:
        return False

    # Re-queue the task to Docket for processing
    task = await self._task_manager.get_task(task_id)
    if task:
        handler = self._create_wrapped_handler()
        async with Docket(url=self._redis_url, name=self._queue_name) as docket:
            queued_func: Any = docket.add(handler, key=task.task_id)
            await queued_func(
                task_id=task.task_id,
                session_id=task.session_id,
                message=task.metadata.get("message", ""),
                context=task.metadata.get("context", {}),
            )

    return True

create_task_with_session

create_task_with_session async

create_task_with_session(task_manager: TaskManager, memory: Memory, message: str, context: dict[str, Any] | None = None, session_id: str | None = None, user_id: str | None = None) -> tuple[TaskState, str]

Create a new task with a memory session.

This is the primary entry point for starting a new conversation. It creates a memory session (if needed), adds the initial message, and creates a task.

Parameters:

Name	Type	Description	Default
task_manager	TaskManager	The TaskManager instance	required
memory	Memory	The Memory instance	required
message	str	The initial user message	required
context	dict[str, Any] | None	Optional context data (user info, instance IDs, etc.)	None
session_id	str | None	Optional existing session ID	None
user_id	str | None	Optional user ID for memory isolation	None

Returns:

Type	Description
tuple[TaskState, str]	Tuple of (TaskState, session_id)

Source code in redis_agent_kit/core.py

async def create_task_with_session(
    task_manager: TaskManager,
    memory: Memory,
    message: str,
    context: dict[str, Any] | None = None,
    session_id: str | None = None,
    user_id: str | None = None,
) -> tuple[TaskState, str]:
    """Create a new task with a memory session.

    This is the primary entry point for starting a new conversation.
    It creates a memory session (if needed), adds the initial message, and creates a task.

    Args:
        task_manager: The TaskManager instance
        memory: The Memory instance
        message: The initial user message
        context: Optional context data (user info, instance IDs, etc.)
        session_id: Optional existing session ID
        user_id: Optional user ID for memory isolation

    Returns:
        Tuple of (TaskState, session_id)
    """
    # Bind memory to session
    bound_memory = memory.with_session(session_id, user_id)

    # Add initial message (creates session if needed)
    await bound_memory.add_message(role="user", content=message)

    # Create task
    task = await task_manager.create_task(
        session_id=bound_memory.session_id,
        message=message,
        context=context,
    )

    # session_id is guaranteed non-None after with_session()
    return task, cast(str, bound_memory.session_id)

RetryConfig

RetryConfig dataclass

RetryConfig(attempts: int = 3, delay: timedelta = timedelta(seconds=5))

Configuration for task retry behavior.

ConcurrencyConfig

ConcurrencyConfig dataclass

ConcurrencyConfig(key: str, max_concurrent: int = 1, scope: str = 'default')

Configuration for task concurrency limits.

Settings

Settings

Bases: BaseSettings

Redis Agent Kit configuration.

Example usage

from redis_agent_kit import Settings, AgentKit

Use defaults

kit = AgentKit()

Use config file

export RAK_CONFIG=/path/to/config.yaml

kit = AgentKit()

Programmatic

settings = Settings(redis_url="redis://prod:6379") kit = AgentKit(settings=settings)