by-framework

by-framework is a distributed, high-performance Agent scheduling engine built on Redis Streams, purpose-built for multi-agent systems.

Challenges in Traditional Architecture

Traditional AI application architectures often face three critical challenges when dealing with Agent scenarios:

• Full-link Synchronous Blocking $\rightarrow$ Forced "Manual Monitoring" — Strong coupling between frontend and backend means tasks are interrupted if the page is closed. Users cannot switch devices or tasks, making workflows fragile to network fluctuations or interruptions.
• Inability to Support Long-running Tasks $\rightarrow$ System "Constant Accompaniment" — For reasoning tasks taking minutes or hours, callers must block threads and wait. This leads to gateway timeouts and massive waste of idle compute resources.
• Inter-Agent Orchestration Recovery Dilemma — In complex cascaded calls, if a timeout or interruption occurs, it's nearly impossible to accurately resume state. Developers are forced to build extremely complex persistent state machines.

The By-Framework Solution

By-Framework addresses these issues through an asynchronous architecture with separated Control and Data Planes:

• Instruction Asynchrony: The APP sends control instructions to the Control Queue via the Gateway Client. Being asynchronous, the APP never blocks, and backend threads are released immediately.
• Agent Cluster Consumption: A distributed cluster of Agents competitively consumes messages from the control queue. Logical routing (Agent Type) provides native load balancing and elastic scaling.
• Data Stream Feedback: During execution, Agents asynchronously push chunks, state changes, and artifacts to the Data Queue. The APP listens via the Gateway Client for progress, natively supporting ultra-long tasks.
• Native Orchestration & Resumption: When an Agent needs to call another Agent, it sends a new instruction to the Control Queue. This message-based mechanism allows tasks to release resources while waiting and resume context precisely upon receiving a reply.

Highlights

• 🔌 Plugin System — Hot-reloadable plugins with lifecycle hooks, tools, prompts, and sub-agent configs
• 🤝 Inter-Agent Orchestration — Built-in call_agent, scatter-gather fan-out, and human-in-the-loop patterns
• 🧩 Extension Ecosystem — Drop-in packages for Langfuse, Phoenix, PostgreSQL, LangGraph, and Google ADK
• 🛡️ Production-Ready — Competitive consumption, graceful shutdown, message persistence, and execution state tracking

Table of Contents

• Architecture
  • Data Flow
  • Component Hierarchy
  • Worker Routing
  • Redis Key Map
• Getting Started
  • Installation
  • Quick Start
• Core Concepts
  • GatewayWorker
  • AgentContext
  • Protocol & Messages
  • Plugin System
• Advanced Features
  • Inter-Agent Calling
  • Scatter-Gather Dispatch
  • User-in-the-Loop
  • Service Discovery
• Sending Tasks
• Extension Libraries
• Configuration Reference
• Development
• Deployment
• Roadmap
• Contributing
• License

Architecture

The system is fully asynchronous and event-driven. Control messages and data events travel on separate Redis Streams, so scaling up Workers does not couple to the data delivery path.

Data Flow

Client ──▶ Redis Control Stream ──▶ GatewayWorker (competitive consume)
               (queue:ctrl:{agent_type})
                                            │
                                            ▼
Consumer ◀──── Redis Data Stream ◀──── Business Logic (emit chunks/states/artifacts)
Backend       (queue:data:{session})

1. A client writes a command to the agent-type control stream.
2. Any online GatewayWorker subscribing to that agent type competitively pulls the message via Redis consumer groups.
3. The Worker processes the task and, through AgentContext, emits streaming chunks, state changes, and artifacts back to a session-scoped data stream.
4. A backend or frontend consumer reads that data stream in real time.

Component Hierarchy

Worker Routing

Three semantic layers govern how tasks reach Workers:

Production path (agent type routing):

• Client writes to byai_gateway:ctrl:agent_type:{agent_type}.
• Multiple Workers in the same consumer group compete for messages.
• Sender only verifies that at least one online Worker exists for the target agent type.

Debug path (direct worker routing):

• When target_worker_id is explicitly provided, the message goes to byai_gateway:ctrl:worker:{worker_id}.
• The sender explicitly checks that the target Worker is online.

Redis Key Map

Getting Started

Installation

Prerequisites: Python 3.12+, Redis 7.0+

# via pip
pip install by-framework

Optional extension packages:

pip install by-framework-trace-langfuse       # Langfuse observability
pip install by-framework-history-postgres     # PostgreSQL history backend
pip install by-framework-langgraph            # LangGraph integration

Quick Start

1. Define a Worker:

# my_agent.py
from by_framework import GatewayWorker, AgentContext, run_worker

class MyAgent(GatewayWorker):
    def get_agent_types(self):
        return ["my_agent"]

    async def process_command(self, command, context: AgentContext):
        await context.emit_chunk("Hello from your agent!")
        return {"status": "completed", "content": "Hello from your agent!"}

if __name__ == "__main__":
    run_worker(MyAgent, worker_id="worker-01")

2. Start Redis:

docker run -d -p 6379:6379 redis:7-alpine

3. Start the Worker:

uv run python my_agent.py

4. Send a task:

# send_task.py
import asyncio
from by_framework import ByaiGatewayClient, WorkerRegistry, init_redis, close_redis

async def main():
    redis = init_redis(host="localhost", port=6379)
    registry = WorkerRegistry(redis_client=redis)
    client = ByaiGatewayClient(redis_client=redis, registry=registry)

    resp = await client.send_message(
        target_agent_type="my_agent",
        session_id="demo-session",
        content="Hello!",
    )
    print(f"Sent: {resp.message_id}")

    await close_redis()

asyncio.run(main())

A runnable version of this pair lives at examples/echo_worker.py (Worker) and examples/send_and_verify.py (client — blocks until it sees the echoed reply) — the same pair deploy/'s CI smoke test drives end to end, see docs/architecture/production-deployment.md.

Core Concepts

GatewayWorker

The abstract base class for all Workers. You implement two methods:

The base class handles, transparently:

• Message lifecycle (parse, decode, acknowledge, persist)
• Workspace provisioning per session/task
• Plugin hook execution (on_task_start, on_task_complete, etc.)
• Sub-agent call orchestration (suspend, resume, cascade cancel)
• History persistence (in-memory or pluggable backend)

AgentContext

The per-task runtime context. Available as the second argument to process_command().

async def process_command(self, command, context: AgentContext):
    # Streaming output
    await context.emit_chunk("Step 1 complete\n")

    # State transitions
    await context.emit_state("analyzing")

    # Artifacts / structured data
    await context.emit_artifact(ArtifactEvent(url="https://example.com/output.json"))

    # Call another agent (with optional suspend-and-wait)
    reply = await context.call_agent(
        target_agent_type="translator",
        content="Hello world",
        wait_for_reply=True,
    )

    # Scatter-gather fan-out
    group = await context.dispatch_group([
        {"target_agent_type": "researcher", "content": "Find references"},
        {"target_agent_type": "writer",    "content": "Draft summary"},
    ])
    results = await context.collect_group_results(group["task_group_id"])

    # Ask the end-user a question (suspends the task)
    return await context.ask_user(AskUserEvent(prompt="Approve deployment?"))

Key properties: session_id, trace_id, message_id, parent_message_id, current_agent_id.

Protocol & Messages

Commands and events are defined in core/protocol/. The system supports these command types:

Event types emitted to data streams:

Message context is carried by MessageHeader (message_id, session_id, trace_id, source_agent_type, target_agent_type, parent_message_id, task_group_id, user_code).

Plugin System

Plugins are the primary extensibility mechanism. They register AgentConfigs that declare tools, prompts, skills, callbacks, and sub-agents — and they hook into the Worker lifecycle.

Writing a Plugin

from by_framework import Plugin, PluginManifest, AgentConfig, PluginBuildContext, AgentContext

class WeatherPlugin(Plugin):
    def __init__(self):
        super().__init__(PluginManifest(plugin_id="weather", version="1.0.0"))

    async def register_agent_configs(self, ctx: PluginBuildContext) -> list[AgentConfig]:
        return [
            AgentConfig(
                agent_id="weather_agent",
                tools={
                    "get_weather": self._get_weather,
                },
                prompts={
                    "system": "You are a weather assistant."
                },
            )
        ]

    async def _get_weather(self, city: str) -> dict:
        return {"city": city, "temp": 22, "condition": "sunny"}

    # Lifecycle hooks
    async def on_task_start(self, context: AgentContext): ...
    async def on_task_complete(self, context: AgentContext, result): ...
    async def on_task_error(self, context: AgentContext, error: Exception): ...
    async def on_task_cancel(self, context: AgentContext, reason: str): ...
    async def on_call_agent_start(self, context: AgentContext, target: str, content): ...
    async def on_call_agent_complete(self, context: AgentContext, target: str, result): ...
    async def on_worker_startup(self): ...
    async def on_worker_shutdown(self): ...

Loading Plugins

Three ways to provide plugins to run_worker():

# 1. Explicit list
run_worker(MyAgent, plugin_list=[WeatherPlugin()])

# 2. Configurator callback
def setup(registry):
    registry.register_bundle(WeatherPlugin())
run_worker(MyAgent, plugin_configurator=setup)

# 3. Directory scan (startup-time)
run_worker(MyAgent, plugin_dir="./my_plugins")

Plugin Hot-Reload

Send a ReloadPluginsCommand to trigger reload() on all plugins without restarting the Worker process. Config snapshots are versioned and persisted to Redis so that even during a restart, the last known-good configuration is recovered.

Advanced Features

Inter-Agent Calling

A Worker can delegate to another agent via context.call_agent(). When wait_for_reply=True, the current task suspends, the callee runs to completion, and the reply is delivered back to the caller as an AgentTaskResult.

The framework handles:

• Task tree construction (parent/child linking for cascade cancel)
• Callback notification when the callee finishes
• Automatic re-delivery of the reply message

Scatter-Gather Dispatch

Fan out multiple sub-tasks in parallel and collect their results:

group = await context.dispatch_group([
    {"target_agent_type": "researcher", "content": "Find papers"},
    {"target_agent_type": "analyst",    "content": "Summarize findings"},
], wait_for_reply=True)

results = await context.collect_group_results(group["task_group_id"])
for r in results.values():
    print(r["content"])

Group progress is tracked in Redis; dispatch_group returns immediately with a task_group_id that can be polled via collect_group_results.

User-in-the-Loop

Suspend a task and wait for human input:

from by_framework import AskUserEvent, ResumeCommand

async def process_command(self, command, context: AgentContext):
    if isinstance(command, ResumeCommand):
        # This is the user's reply
        await context.emit_chunk(f"You said: {command.content}")
        return {"status": "completed"}

    # Suspend and ask
    return await context.ask_user(
        AskUserEvent(prompt="What is the target deployment environment?")
    )

Service Discovery

Redis-backed service discovery utilities:

Sending Tasks

GatewayClient

from by_framework import GatewayClient, WorkerRegistry, init_redis

redis = init_redis(host="localhost", port=6379)
client = GatewayClient(redis_client=redis, registry=WorkerRegistry(redis_client=redis))

# Send to an agent type
resp = await client.send_message(
    target_agent_type="my_agent",
    session_id="sess-001",
    content="Your task content",
    user_code="user-123",
    metadata={"priority": "high"},
)

# Cancel a task (including cascade through sub-tasks)
await client.cancel_task(
    message_id=resp.message_id,
    session_id="sess-001",
    reason="User requested",
)

ByaiGatewayClient

A typed wrapper around GatewayClient that automatically serializes content through the Byai codec:

from by_framework import ByaiGatewayClient, BaiYingMessage

client = ByaiGatewayClient(redis_client=redis, registry=registry)

# Automatically encodes BaiYingMessage content
resp = await client.send_message(
    target_agent_type="chat_agent",
    session_id="sess-001",
    content=BaiYingMessage(role="user", content="Hello"),
)

Interceptors

Register request interceptors on the client side for custom pre-processing:

class AuthInterceptor:
    async def before_send(self, command, header):
        header.metadata["auth_token"] = "..."
        return command, header

client = GatewayClient(...)
client.add_interceptor(AuthInterceptor())

Extension Libraries

These are optional workspace member packages shipping alongside the core framework:

Tracing providers are auto-discovered via TraceProviderFactory at Worker startup when the corresponding package is installed and environment variables are configured.

Configuration Reference

run_worker() Parameters

Environment Variables

Development

# Install all workspace dependencies
make install

# Format (isort + ruff + pyink)
make format

# Lint (pylint + ruff)
make lint

# Run all tests
make test

# Run a single test file
uv run pytest tests/worker/test_gateway_worker.py

# Run tests matching a pattern
uv run pytest -k "test_name_pattern"

# Full CI check
make ci

Code style: isort for imports, ruff-format + pyink for formatting, pylint + ruff for linting. Pre-commit hooks run automatically on git commit.

Tests are organized by module under tests/:

• tests/common/ — Logger, Redis client, config, exceptions
• tests/core/ — Registry, protocol, discovery
• tests/worker/ — Worker, runner, context, processor, emitter, sandbox
• tests/client/ — Client functionality
• tests/plugin/ — Plugin registry, system, discovery, tracing
• tests/integration/ — End-to-end flows (scatter-gather, callbacks, ask_user)

Deployment

Single Machine

# 1. Start Redis
docker run -d --name by-redis -p 6379:6379 redis:7-alpine

# 2. Start a Worker
python -m by_framework \
  --worker-class my_agent.MyAgent \
  --worker-id worker-01 \
  --redis-host localhost

Horizontal Scaling

Run multiple Worker processes with different worker_id values. They all consume from the same agent-type control stream. Redis consumer groups automatically handle load distribution.

python -m by_framework --worker-class my_agent.MyAgent --worker-id worker-01 &
python -m by_framework --worker-class my_agent.MyAgent --worker-id worker-02 &
python -m by_framework --worker-class my_agent.MyAgent --worker-id worker-03 &

& is fine for a quick local check but isn't real process supervision — for an actual deployment, use the reference Dockerfile, Compose file, and Kubernetes Deployment under deploy/:

# Docker Compose — N supervised replicas, distinct worker ids, restart policy
docker compose -f deploy/docker-compose.yml up --build --scale worker=3

# Kubernetes — same idea as a Deployment
kubectl apply -f deploy/kubernetes/worker-deployment.yaml

The CLI also exposes --redis-password, --redis-username, --redis-mode, and --redis-cluster-nodes (all mirrored by the REDIS_* env vars used in the Compose/Kubernetes examples above) — see python -m by_framework --help.

Reliability

• Message persistence: Messages are stored in Redis Streams until explicitly acknowledged (XACK). Unacknowledged messages are redelivered on Worker restart.
• Durable config: Agent config snapshots are persisted to Redis, so a restarted Worker recovers the last-known plugin configuration.
• Graceful shutdown: WorkerRunner drains in-flight tasks before shutting down, acknowledging completed work. This is wired to both SIGINT (Ctrl+C) and SIGTERM — the signal docker stop/Kubernetes pod termination actually send — so it fires under real container orchestration, not just an interactive terminal. Give it enough stop_grace_period/terminationGracePeriodSeconds to cover your longest task (see the deploy/ examples), since a hard SIGKILL after the grace period skips the drain entirely.
• Separate data path: Data output goes to session-scoped streams independently of control, so backend consumers are decoupled from Worker scaling.

Logging

from by_framework.common.logger import setup_logging
import logging

setup_logging(level=logging.INFO, use_json=True)  # JSON for log aggregation

Observability Dashboard

Serve the built-in dashboard to inspect worker health, agent health, execution state counts, recent executions, Redis stream queue depth, consumer-group pending/lag, failure details, routing decisions, derived alerts, and segmented queue/run/end-to-end task latency:

uv run --package by-framework-dashboard by-framework-dashboard --host 127.0.0.1 --port 8765

Open http://127.0.0.1:8765/. For a local UI preview without Redis, open http://127.0.0.1:8765/?demo=1. Prometheus-style metrics are available at http://127.0.0.1:8765/metrics, and the dashboard keeps short in-memory trend history at http://127.0.0.1:8765/api/history. The UI uses split polling endpoints (/api/workers, /api/executions, /api/queues, /api/history) instead of polling the full /api/snapshot endpoint on every refresh. Runtime self-check data is exposed at /api/health, shown in the toolbar, and exported through /metrics. Alert thresholds can be tuned with --queue-backlog-threshold, --delivery-pending-threshold, --consumer-pending-threshold, and --failed-execution-threshold.

The dashboard frontend is built with React/Vite under libs/by-framework-dashboard/frontend; its production build is packaged in libs/by-framework-dashboard/src/by_framework_dashboard/static. See libs/by-framework-dashboard/README.md for dashboard-specific runbook details.

Roadmap

• Observability dashboard for Worker health and task streams
• WASM-based sandbox for stronger execution isolation
• Enhanced LangGraph multi-agent orchestration adapter

Contributing

Issues and pull requests are welcome. See CONTRIBUTING.md for guidelines.

License

Apache 2.0 — see LICENSE.

Maintained by the byai team.