Flux Agents¶

Up and Running in 30 Seconds¶

Install, import, and get a response from any supported provider.

InstallCodeAsyncStreaming

pip install flux-agents

from flux import Agent, Runner

agent = Agent(name="assistant", instructions="You are helpful")
result = Runner.run_sync(agent, "Hello!")
print(result.final_output)

import asyncio
from flux import Agent, Runner

async def main():
    agent = Agent(name="assistant", instructions="You are helpful")
    result = await Runner.run(agent, "Hello!")
    print(result.final_output)

asyncio.run(main())

import asyncio
from flux import Agent, Runner

async def main():
    agent = Agent(name="assistant", instructions="Tell me a joke")
    stream = await Runner.run_streamed(agent, "Tell me a joke")
    async for event in stream:
        if hasattr(event, "delta"):
            print(event.delta, end="", flush=True)
    print()

asyncio.run(main())

Everything You Need to Build Agentic Systems¶

Flux gives you a complete toolkit for building, testing, and deploying AI agents -- from single-purpose assistants to complex multi-agent pipelines.

Provider Agnostic

Switch between Ollama, OpenAI, Anthropic, Groq, DeepSeek, and OpenRouter with a single import change. No vendor lock-in, ever.

Providers
Async First

Non-blocking by default for maximum throughput. Need synchronous code? Runner.run_sync() is always available.

Runners
Protocol Based

Structural typing via Protocol -- no inheritance required. Any class with the right methods is a valid Tool, Model, or Session.

Protocols
Middleware

Composable middleware for logging, caching, retry, and rate-limiting. Wrap any agent run with zero coupling to the core framework.

Middleware
Event Driven

Decoupled event bus for observability and analytics. Subscribe to agent, tool, and session events without modifying the runner.

Events
Tools

Decorate any function with @tool and it becomes callable by agents. Built-in tools for shell, file I/O, and more.

Tools
Handoffs

Agent-to-agent routing for multi-agent systems. Let a router delegate to specialist agents automatically.

Handoffs
Guardrails

Input and output validation built in. Length checks, PII detection, profanity filtering, and custom guardrails.

Guardrails
Sessions

In-memory sessions for ephemeral chats and SQLite-backed sessions for persistence across restarts.

Sessions
Memory

Conversation memory and vector-based retrieval for long-term knowledge across sessions.

Memory
Streaming

Real-time token streaming with structured events. Get partial responses as they are generated.

Streaming
Tracing

Console and file-based tracing for debugging and auditing. Capture every model call, tool invocation, and handoff.

Tracing

Architecture at a Glance¶

A clean, layered design where every component is optional and composable.

graph TB
    subgraph User["User Application"]
        U[Your Code]
    end

    subgraph Flux["Flux Framework"]
        R[Runner]
        A[Agent]
        MW[Middleware Stack]
        EB[Event Bus]
        GR[Guardrails]
    end

    subgraph Providers["LLM Providers"]
        OLL[Ollama]
        OAI[OpenAI]
        ANT[Anthropic]
        GROQ[Groq]
        DS[DeepSeek]
        OR[OpenRouter]
    end

    subgraph Storage["Persistence & Memory"]
        IM[InMemory Session]
        SQL[SQLite Session]
        VM[Vector Memory]
    end

    subgraph Ext["Extensions"]
        T[Tools]
        HO[Handoffs]
        TR[Tracing]
    end

    U --> R
    R --> A
    A --> MW
    MW --> R
    R --> GR
    R --> Providers
    R --> EB
    R --> T
    R --> HO
    R --> Storage
    EB --> TR

    style User fill:#f3e5f5,stroke:#7b1fa2,stroke-width:2px
    style Flux fill:#fff3e0,stroke:#f57c00,stroke-width:2px
    style Providers fill:#e8f5e9,stroke:#388e3c,stroke-width:2px
    style Storage fill:#e3f2fd,stroke:#1565c0,stroke-width:2px
    style Ext fill:#fce4ec,stroke:#c62828,stroke-width:2px

Why Flux?¶

Six design principles that set Flux apart.

Protocol over ABC

No inheritance required. Any class with the right methods works as a Model, Tool, or Session. Your code stays yours -- no framework coupling.

# Works immediately -- no import, no base class
class MyModel:
    async def complete(self, request): ...
    async def stream(self, request): ...

Async First

Every code path is async for maximum concurrency. Sync wrappers are provided for scripts and notebooks, so you never pay for what you don't use.
```
# Async (default)
result = await Runner.run(agent, "Hello!")
# Sync wrapper
result = Runner.run_sync(agent, "Hello!")
```
Zero Core Dependencies

The base framework ships with zero third-party dependencies. Provider packages are installed on demand -- no bloated environments.
```
# Minimal install
pip install flux-agents
# Add only what you need
pip install flux-agents[ollama]
pip install flux-agents[openai]
```

Middleware over Hooks

Composable middleware wraps the entire run cycle. Modify requests, log calls, add caching, or retry failures -- all without touching the runner.

from flux import RetryMiddleware, CacheMiddleware

agent = Agent(
    name="cached",
    instructions="Be helpful",
    middleware=[CacheMiddleware(ttl=300), RetryMiddleware(max_retries=3)],
)

Immutable Agents

Agents are immutable dataclasses. Share them across threads without fear. Use clone() to create modified copies safely.
```
agent = Agent(name="bot", instructions="Helpful")
specialist = agent.clone(name="specialist", instructions="Expert mode")
```
:material-broadcast-outline:{ .lg .middle } Event-Driven

A decoupled event bus for observability, analytics, and custom hooks. Subscribe from anywhere -- no framework modification needed.
```
from flux import get_event_bus

bus = get_event_bus()
bus.on("agent.start", lambda e: print(f"Started: {e.data}"))
bus.on("tool.end", lambda e: log_to_analytics(e))
```

Installation¶

One command to start. Add providers as you need them.

CoreOllamaOpenAIAnthropicAll Providers

pip install flux-agents

Zero Dependencies

The core package has no third-party dependencies. Provider SDKs are installed separately.

pip install flux-agents[ollama]

from flux.models.ollama import OllamaModel

model = OllamaModel(model="qwen2:1.5b")
agent = Agent(name="local", instructions="Be helpful", model=model)

pip install flux-agents[openai]

from flux.models.openai_provider import OpenAIModel

model = OpenAIModel(model="gpt-4o-mini")
agent = Agent(name="gpt", instructions="Be helpful", model=model)

pip install flux-agents[anthropic]

from flux.models.anthropic import AnthropicModel

model = AnthropicModel(model="claude-sonnet-4-20250514")
agent = Agent(name="claude", instructions="Be helpful", model=model)

pip install flux-agents[full]

See It in Action¶

Real patterns you can copy into your projects.

ToolsHandoffsGuardrailsSessionsMiddlewareCustom Model

from flux import Agent, Runner, tool

@tool
def calculator(expression: str) -> str:
    """Evaluate a math expression."""
    return str(eval(expression))

agent = Agent(
    name="math_bot",
    instructions="Use calculator for math",
    tools=[calculator],
)

result = Runner.run_sync(agent, "What is 15 * 23?")
print(result.final_output)

import asyncio
from flux import Agent, Runner
from flux.handoffs.handoff import Handoff

async def main():
    coder = Agent(name="coder", instructions="Write code")
    writer = Agent(name="writer", instructions="Write content")

    router = Agent(
        name="router",
        instructions="Route to specialist",
        handoffs=(
            Handoff(source=router, target=coder),
            Handoff(source=router, target=writer),
        ),
    )

    result = await Runner.run(router, "Write Python hello world")
    print(f"Handled by: {result.last_agent.name}")

asyncio.run(main())

from flux import Agent, Runner, LengthGuardrail, PIIGuardrail

agent = Agent(
    name="safe_bot",
    instructions="Be helpful",
    guardrails=(
        LengthGuardrail(max_chars=5000),
        PIIGuardrail(),
    ),
)

result = Runner.run_sync(agent, "Tell me about Python")

import asyncio
from flux import Agent, Runner, SQLiteSession

async def main():
    agent = Agent(name="bot", instructions="Remember everything")
    session = SQLiteSession(db_path="chat.db")

    r1 = await Runner.run(agent, "My name is Sara", session=session)
    r2 = await Runner.run(agent, "What is my name?", session=session)
    print(f"Bot: {r2.final_output}")

asyncio.run(main())

from flux import Agent
from flux.middleware.base import Middleware, NextFn, RequestContext, Response

class TimingMiddleware:
    async def process(self, ctx: RequestContext, next: NextFn) -> Response:
        import time
        start = time.time()
        response = await next(ctx)
        print(f"Elapsed: {time.time() - start:.2f}s")
        return response

agent = Agent(
    name="timed",
    instructions="Be helpful",
    middleware=[TimingMiddleware()],
)

from flux import Agent, Runner
from flux.models.base import ModelRequest, ModelResponse, StreamChunk

class MyModel:
    async def complete(self, request: ModelRequest) -> ModelResponse:
        return ModelResponse(content="Hello from my model!")

    async def stream(self, request: ModelRequest):
        yield StreamChunk(delta_text="Hello!", done=True)

agent = Agent(name="custom", instructions="Be helpful", model=MyModel())
result = Runner.run_sync(agent, "Hi")
print(result.final_output)

Explore the Documentation¶

Getting Started

Installation, quickstart, and your first agent.

Getting Started
Core Concepts

Agents, tools, providers, sessions, middleware, and more.

Core Concepts
API Reference

Complete class and function documentation.

API Reference
Guides

Step-by-step tutorials for real-world patterns.

Guides
Examples

Copy-paste code samples for every feature.

Examples
Architecture

Design decisions, directory structure, and execution flow.

Architecture

Community¶

GitHub

Source code, issue tracker, and release notes. Contributions are welcome.

flux-agents/flux
:material-pypi:{ .lg .middle } PyPI

Published packages for core and all provider extras.

flux-agents on PyPI
Twitter / X

Announcements, tips, and community highlights.

@flux_agents

Roadmap¶

What is coming next for Flux Agents.

Upcoming Features

v0.2 -- Structured Output & Pydantic Integration

Define output schemas with Pydantic models. Agents will return validated, typed data instead of raw strings.

v0.2 -- Agent Cloning with Presets

Pre-built agent templates for common patterns: chatbot, researcher, coder, and summarizer.

v0.3 -- HTTP Server & REST API

Built-in HTTP server to expose agents as API endpoints with streaming support.

v0.3 -- MCP (Model Context Protocol) Support

Connect Flux agents to MCP servers for tool and resource discovery across the ecosystem.

v0.4 -- Evaluation & Benchmarking

Built-in evaluation harness to test agent accuracy, latency, and cost across providers.

v0.4 -- Multi-Modal Support

Vision and audio inputs through provider-specific multi-modal APIs.

Get Involved

Flux is open source and community-driven. Check the Contributing guide to get started, or open an issue on GitHub to request a feature.

Ready to build your first agent?¶

Get Started Now