Tasks¶

A Task represents a unit of agent work—typically one turn in a conversation. Tasks track status, progress updates, results, and errors.

Creating Tasks¶

from redis_agent_kit import TaskManager
import redis.asyncio as redis

client = redis.from_url("redis://localhost:6379", decode_responses=True)
task_manager = TaskManager(client)

# Create a task linked to a session
task = await task_manager.create_task(
    session_id="session_123",
    message="What is Redis?",
)
print(task.task_id)              # "01HXYZ..."
print(task.status)               # TaskStatus.QUEUED
print(task.metadata["message"])  # "What is Redis?"

Task Lifecycle¶

Tasks progress through these statuses:

QUEUED → IN_PROGRESS → DONE
                     → FAILED
                     → CANCELLED
         ↓
    AWAITING_INPUT    → QUEUED (after submit_input)
    AWAITING_CHILDREN → QUEUED (after last child terminates)

AWAITING_INPUT pauses the task for a user response (see Input Handling). AWAITING_CHILDREN pauses the task on spawned sub-tasks (see Sub-Tasks). Both unblock by re-enqueueing the task on the original queue.

Updating Tasks¶

Use update() to change status, add messages, set results, or set errors—in one call:

# Start processing with a message
await task_manager.update(task.task_id, status="in_progress", message="Starting...")

# Add progress updates
await task_manager.update(task.task_id, message="Found 5 documents")
await task_manager.update(task.task_id, message="Calling API", update_type="tool_call")

# Complete with result
await task_manager.update(task.task_id, status="done", result={"answer": "Redis is..."})

# Or fail with error
await task_manager.update(task.task_id, status="failed", error="API timeout")

Update types: info (default), tool_call, reflection, error

You can also use the individual convenience methods:

await task_manager.update_status(task.task_id, TaskStatus.IN_PROGRESS)
await task_manager.add_update(task.task_id, "Processing...")
await task_manager.set_result(task.task_id, {"answer": "..."})
await task_manager.set_error(task.task_id, "Something went wrong")

Retrieving Tasks¶

from redis_agent_kit import TaskStatus

# Get a single task
task = await task_manager.get_task(task_id)
print(task.status, task.result, task.updates)

# List tasks by status
tasks = await task_manager.list_tasks(status=TaskStatus.IN_PROGRESS)
for t in tasks:
    print(f"{t.task_id}: {t.metadata.get('message')}")

# List tasks for a session
tasks = await task_manager.list_tasks(session_id=session_id)

TaskEmitter¶

For convenience, use TaskEmitter to emit updates without passing task_id everywhere:

from redis_agent_kit import TaskEmitter

emitter = TaskEmitter(task_manager, task.task_id)

# Fire-and-forget
emitter.emit_nowait("Processing query...")

# Awaitable
await emitter.emit("Analysis complete", metadata={"tokens": 150})

# Stream a single LLM token (ephemeral, not persisted)
await emitter.emit_token("Hello")

Watching Progress¶

Updates are persisted to the task and can be polled via get_task() or streamed live via Server-Sent Events when streaming is enabled:

from redis_agent_kit import AgentKit, StreamConfig

kit = AgentKit("redis://localhost:6379", agent_callable=my_handler, stream_config=StreamConfig(enabled=True))

Clients then connect to GET /tasks/{task_id}/stream to receive update, token, done, failed, and cancelled events in real time. See the Streaming guide for channel scopes, event filtering, and token streaming.

Deleting Tasks¶

await task_manager.delete_task(task_id)

Custom Prefix¶

All keys use rak: by default. Use a custom prefix:

task_manager = TaskManager(client, prefix="myapp")
# Keys: myapp:task:{id}, myapp:tasks:status:{status}

Task Fields¶

Field	Type	Description
`task_id`	str	Unique ID (ULID)
`session_id`	str	Parent session ID
`status`	TaskStatus	Current status
`result`	dict	Result data
`error_message`	str	Error message
`updates`	list[TaskUpdate]	Progress updates
`metadata`	dict	Arbitrary metadata (includes `message`)
`input_request`	InputRequest	Pending input request
`input_response`	dict	User's input response
`parent_id`	str \| None	Parent task ID (sub-tasking)
`spawn_key`	str \| None	Stable label set by parent at spawn (sub-tasking)
`child_ids`	list[str]	Spawned children (sub-tasking)
`pending_children`	int	Non-terminal child count (sub-tasking)
`depth`	int	Nesting depth — 0 for root (sub-tasking)
`created_at`	datetime	Creation time
`updated_at`	datetime	Last update time