flowra 0.0.7

Flowra — flow infrastructure for building stateful LLM agents

Flowra

Flow infra for building stateful, persistent LLM agents with tool use, parallel execution, and crash recovery. Requires Python 3.12+.

Features

• State machine agents — define agents as Agent[Spec, Result] classes with @step methods, a single entry point, and typed spec/result contracts
• Persistent state — Scalar[T] and AppendOnlyList[T] with incremental dirty-tracking and pluggable storage (in-memory, file-based, or custom)
• Tool integration — @tool decorator for local functions, MCP server support, DI into tool handlers, agents as tools for LLM-driven delegation
• LLM abstraction — provider-agnostic LLMProvider interface with immutable message types and real-time streaming (ships AnthropicVertexProvider, GoogleVertexProvider, OpenAIProvider)
• Agents as tools — @agent_tool decorator exposes an agent as a tool the LLM can call autonomously; sub-agent runs its own system prompt and tool loop
• Cooperative interrupts — InterruptToken for graceful cancellation across the entire execution tree
• Pre-built agents — ChatAgent (multi-turn chat with session history) and ToolLoopAgent (single-turn LLM tool loop with hooks and caching)

Installation

# Base package (no LLM providers)
pip install flowra

# With specific providers
pip install flowra[anthropic]
pip install flowra[openai]
pip install flowra[google]

# All providers
pip install flowra[all]

Quick start

import asyncio

from flowra.lib.chat import ChatAgent, ChatConfig, ChatResult, ChatSpec
from flowra.lib import LLMConfig
from flowra.llm import LLMProvider, SystemMessage, TextBlock
from flowra.llm.providers.anthropic_vertex import AnthropicVertexProvider
from flowra.runtime import AgentRuntime
from flowra.tools import ToolRegistry


async def main() -> None:
    provider = AnthropicVertexProvider()

    async with await ToolRegistry.create([]) as registry:
        config = ChatConfig(
            llm_config=LLMConfig(model="claude-sonnet-4-5@20250929"),
            system_messages=[
                SystemMessage(blocks=[TextBlock(text="You are a helpful assistant.")])
            ],
        )

        runtime = AgentRuntime(
            agents={"chat": ChatAgent},
            services={
                LLMProvider: provider,
                ToolRegistry: registry,
                ChatConfig: config,
            },
        )

        while True:
            user_input = input("You: ")
            if not user_input:
                break

            result = await runtime.run(
                agent=ChatAgent,
                spec=ChatSpec(user_message=user_input),
            )

            if isinstance(result, ChatResult) and result.response:
                print(f"Assistant: {result.response}")


asyncio.run(main())

Package structure

flowra/
├── llm/        # LLM abstraction (messages, blocks, provider interface)
├── tools/      # Tool definition, registration, execution
├── agent/      # State machine framework (@step, Agent[S,R], stored values)
├── runtime/    # Execution engine, persistence, interrupt support
└── lib/        # Pre-built agents (ChatAgent, ToolLoopAgent, hooks, caching)

See docs/architecture.md for the full dependency graph and data flow. Each package has its own documentation in docs/.

Development

make deps      # install dependencies (uv sync)
make check     # lint + test
make chat      # run interactive console chat example

Documentation

• Architecture — package structure and data flow
• LLM — message types, provider interface
• Tools — tool definition and execution
• Agent — state machine framework
• Runtime — execution engine and persistence
• Lib — pre-built agents, hooks, caching