fsm-agent-flow v0.3.0

Write LLM workflows
that test themselves.

Every state declares what it must accomplish. The framework validates before advancing. Failed states retry with feedback. Like TDD for your agent pipeline.

start
research
Gather info + cite sources
pass
writing
Structured report w/ headings
pass
end
review
Quality check
Each node validates its key results before the arrow fires. Failures retry with feedback.

Why this approach works

ConceptWhat it means
StateSpecA node in your workflow. Has an objective, tools, an execute function, and acceptance criteria (key results).
KeyResultA verifiable deliverable. Can be a programmatic check (lambda) or an LLM-evaluated description. The "test" in TDD.
TransitionHow states connect. Static (str), conditional (dict with branches), or dynamic (callable). Supports loops and bidirectional flows.
BoundLLMLLM adapter scoped to a state's tools. ctx.llm.run_with_tools() handles the full tool-call loop.
SharedContextKey-value store shared across all states. No globals, no singletons.
ValidatorChecks key results after execution. Rule-based (code), LLM-based (subjective), or custom protocol.

Retry with Feedback

When a state fails validation, it re-executes with ctx.feedback explaining exactly what went wrong. The LLM self-corrects.

Bidirectional Flow

States can branch, loop back, call each other. Return {"_transition": "branch"} to route conditionally at runtime.

Zero Dependencies

No required runtime deps. Bring your own LLM client. OpenAI and LiteLLM adapters ship as optional extras.

01

Install the framework

One command. Pick an LLM adapter or bring your own.

bash
pip install fsm-agent-flow[openai]
Using uv?   uv add fsm-agent-flow[openai] works too. For LiteLLM: pip install fsm-agent-flow[litellm]
02

Build your first workflow

Two states. Research gathers info, writing produces a report. Each state declares what "done" looks like.

python
from fsm_agent_flow import Workflow, StateSpec, KeyResult
from fsm_agent_flow.llm.openai import OpenAIAdapter

# Tools are plain functions — auto-converted to JSON Schema
def search(query: str) -> str:
    """Search the web for information."""
    return your_search_api(query)

# Each state declares its acceptance criteria
research = StateSpec(
    name="research",
    objective="Gather information on the topic",
    key_results=[
        KeyResult("length", "At least 200 chars",
                  check=lambda o: len(str(o)) > 200),
        KeyResult("sourced", "Cites real sources"),  # LLM-validated
    ],
    execute=lambda ctx: ctx.llm.run_with_tools(
        "Research the topic thoroughly.", ctx.input),
    tools=[search],
    max_retries=2,
    is_initial=True,
)

writing = StateSpec(
    name="writing",
    objective="Write a structured report",
    key_results=[
        KeyResult("sections", "Has headings",
                  check=lambda o: str(o).count("#") >= 2),
    ],
    execute=lambda ctx: ctx.llm.run_with_tools(
        "Write a structured report.", str(ctx.input)),
    is_final=True,
)

# Wire it up and run
llm = OpenAIAdapter(model="gpt-4o")
result = Workflow(
    objective="Research and report",
    states=[research, writing],
    transitions={"research": "writing"},
    llm=llm, validator_llm=llm,
).run("quantum computing")

print(result.history[-1].output)
What happens:  research executes → key results checked → if "length" fails, retries with feedback → once both pass, output flows to writing → same validation loop → done.
03

Add branching & loops

Real state machines go back and forth. Return {"_transition": "label"} from execute to pick a branch.

start
check
Need weather?
need_weather
fetch
Get weather API
default
wrong_city ↻ self
ready
end
respond
Print answer
python
transitions = {
    "check": {"need_weather": "fetch", "ready": "respond"},
    "fetch": {"wrong_city": "fetch", "default": "check"},
    "respond": None,
}

# In your execute function, pick the branch:
def check(ctx):
    if ctx.shared.get("weather"):
        return {"_transition": "ready", "answer": "Sunny!"}
    return {"_transition": "need_weather"}
04

Design workflows visually

Launch the node editor in your browser. Drag states, connect transitions, set key results. Export as Python or JSON.

bash
python -m fsm_agent_flow.editor
1
Add state nodes

Click + State in the toolbar. Each node becomes a StateSpec.

2
Connect transitions

Drag from an output port to an input port. For branching, add named conditions in the sidebar.

3
Edit properties

Double-click a node. Set objective, tools, key results, max retries. Toggle initial/final.

4
Export

Hit Export .py for runnable Python code, or Save for portable JSON. Ctrl+S works too.

fsm-agent-flow editor
research
Gather info
writing
Write report
Name
research
Objective
Gather info
Tools
search_web
Key Results
2 defined
Retries
3
05

Teach Claude Code your framework

Paste this into your project's CLAUDE.md. Claude Code will understand the full API and write workflows for you.

markdown — paste into your CLAUDE.md
# fsm-agent-flow

TDD/OKR-driven agentic workflow framework. See the reference docs:

@https://raw.githubusercontent.com/NewJerseyStyle/FSM-agent-flow/main/CLAUDE.md
@https://raw.githubusercontent.com/NewJerseyStyle/FSM-agent-flow/main/docs/claude/rules/adapters.md
@https://raw.githubusercontent.com/NewJerseyStyle/FSM-agent-flow/main/docs/claude/rules/workflows.md
@https://raw.githubusercontent.com/NewJerseyStyle/FSM-agent-flow/main/docs/claude/rules/validation.md
@https://raw.githubusercontent.com/NewJerseyStyle/FSM-agent-flow/main/docs/claude/rules/tools.md
What this does: The @ syntax tells Claude Code to fetch and read those files as context. It will learn the full architecture — state specs, transitions, tools, validation, adapters — and can write complete workflows, debug issues, and build custom adapters in your project.

Quick reference

PatternCode
LLM + tools loopctx.llm.run_with_tools(system, user_msg)
Single LLM callctx.llm.chat([Message(role="user", content="...")])
Share datactx.shared.set("key", val) / ctx.shared.get("key")
Conditional routereturn {"_transition": "branch_name", ...}
Nested OODA agentrun_ooda(ctx, task="...", tools=[...], max_cycles=3)
Retry feedbackCheck ctx.feedback — it explains why the last attempt failed
Save workflow statewf.context.to_dict() / WorkflowContext.from_dict(d)
Visual editorpython -m fsm_agent_flow.editor
JSON roundtripworkflow_to_json(wf) / workflow_from_json(data, llm=...)
Generate Pythonworkflow_to_python(json_data)

Start shipping

Read the README, browse the examples, or just pip install and go.

GitHub Repo ↗ README ↗ Examples ↗