redis-agent-kit 0.3.0

Reusable infrastructure for building AI agents with Redis

Redis Agent Kit

Redis Agent Kit is an experimental Python library for building Redis-backed agent services. It provides durable task state, background workers, progress updates, streaming, memory hooks, RAG ingestion, and protocol adapters without requiring a specific LLM framework.

The project is early and under active development. APIs may change while the library settles.

What It Provides

• Task lifecycle: create, queue, cancel, restart, inspect, and delete agent work.
• Background execution: submit tasks from an API process and run them in independently scaled workers.
• Agent context: pass a TaskContext with the user message, session, emitter, memory facade, attachments, and kit managers.
• Progress updates: persist task milestones and optionally stream updates or tokens over Server-Sent Events.
• RAG pipelines: prepare, chunk, embed, store, and search documents in Redis.
• Protocol support: expose agents through REST, A2A, ACP, and MCP.

Installation

pip install "redis-agent-kit[api,cli]"

Optional extras:

pip install "redis-agent-kit[mcp]"       # MCP server
pip install "redis-agent-kit[memory]"    # working/long-term memory (Python 3.12 only)
pip install "redis-agent-kit[examples]"  # example app dependencies
pip install "redis-agent-kit[all]"       # all optional RAK features

Redis Agent Kit requires Python 3.11+ and Redis. Vector search requires Redis Stack or Redis 8.

Memory (agent-memory-server) requires Python 3.12. On other Python versions the memory extra is skipped; either run on 3.12, install with a slimmer extra (e.g. [api,cli]), or set RAK_MEMORY__ENABLED=false.

Quickstart

Start Redis:

docker run -d --name redis -p 6379:6379 redis:8

Create app.py:

from redis_agent_kit import AgentKit, EmitterMiddleware, TaskContext
from redis_agent_kit.api import create_app


async def my_agent(ctx: TaskContext) -> dict:
    await ctx.emitter.emit("Working...")
    return {"answer": f"Processed: {ctx.message}"}


kit = AgentKit(
    "redis://localhost:6379",
    agent_callable=my_agent,
    middleware=[EmitterMiddleware(start_message="Starting...")],
    queue_name="my_agent",
)

# Docket workers load an iterable task collection from module:path.
tasks = [kit.worker_task]

app = create_app(kit=kit)

Run the worker and API server in separate terminals:

rak worker --name my_agent --tasks app:tasks
uvicorn app:app --reload

Submit and inspect a task:

curl -X POST http://localhost:8000/tasks \
  -H "Content-Type: application/json" \
  -d '{"message": "What is Redis?"}'

{"task_id": "01J...", "session_id": "01J...", "status": "queued", "message": "Task created and queued for processing"}

curl http://localhost:8000/tasks/01J...

Task Handlers

New handlers should accept TaskContext:

async def handler(ctx: TaskContext) -> dict:
    await ctx.emitter.emit("Searching...")
    history = await ctx.memory.get_messages(limit=10)
    return {"response": f"{len(history)} previous messages"}

Legacy handlers with (task_id, session_id, message, context) are still accepted, but new code should prefer TaskContext.

Streaming

Use StreamConfig when clients need live task updates:

from redis_agent_kit import AgentKit, StreamConfig
from redis_agent_kit.api import create_app

stream_config = StreamConfig(enabled=True)

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

Clients can connect to the per-task SSE endpoint:

const es = new EventSource(`/tasks/${taskId}/stream`);
es.addEventListener("update", (e) => console.log(JSON.parse(e.data).message));
es.addEventListener("token", (e) => process.stdout.write(JSON.parse(e.data).message));
es.addEventListener("done", (e) => {
  console.log(JSON.parse(e.data).result);
  es.close();
});

Inside a handler, use await ctx.emitter.emit(...) for persisted updates and await ctx.emitter.emit_token(...) for ephemeral token events.

Pipelines

Pipelines are included in the base package:

rak pipelines run ./docs --pattern "*.md" --chunk-size 800
rak pipelines status

The REST API also exposes pipeline endpoints under /pipelines. See the Pipelines guide for current request bodies and staged ingestion details.

Protocols

Use the REST app factory for standard HTTP plus optional A2A and ACP:

from redis_agent_kit import AgentCard, AgentManifest, Skill
from redis_agent_kit.api import create_app

agent_card = AgentCard(
    name="My Agent",
    description="Answers questions",
    url="http://localhost:8000",
    skills=[Skill(id="chat", name="Chat", description="General chat")],
)

agent_manifest = AgentManifest(
    name="my-agent",
    description="Answers questions",
)

app = create_app(
    kit=kit,
    enable_a2a=True,
    enable_acp=True,
    agent_card=agent_card,
    agent_manifest=agent_manifest,
)

Discovery endpoints:

• A2A: GET /.well-known/agent.json
• ACP: GET /agents
• REST/OpenAPI: GET /docs

Documentation

• Tutorial
• Tasks
• Sessions
• Memory
• Streaming
• Middleware
• Protocols
• Input Handling
• Sub-Tasks
• Pipelines
• CLI
• API

License

MIT