Build AI Agents with Python Decorators, Not Boilerplate
Install • Quick Start • Features • How It Works • Docs
FlowForge is a Python framework that lets you define complex AI agent pipelines using decorators alone. No graph construction code, no YAML configs, no LangChain ceremony — just annotate your classes and functions, and FlowForge compiles them into an executable DAG.
Current version: 0.1.1
from flowforge import global_config, flow, task, step, FlowForge
@global_config(prompt="You are a helpful research assistant.")
class MyAgent:
@flow(name="research", prompt="Analyze query and find answers")
class ResearchFlow:
@task(name="search", prompt="Search for relevant information")
class SearchTask:
@step(order=1, prompt="Expand and optimize the query")
async def expand_query(ctx):
return {"query": ctx.input["query"], "expanded": True}
@step(order=2, prompt="Fetch results from sources")
async def fetch_results(ctx):
return {"results": ["..."], "source": "web"}
engine = FlowForge.compile(MyAgent)
result = await engine.run({"query": "AI agent frameworks in 2026"})pip install flowforgeWith optional extras:
pip install "flowforge[viz]" # Graphviz rendering
pip install "flowforge[openai]" # OpenAI provider support
pip install "flowforge[google]" # Gemini provider support
pip install "flowforge[mcp]" # MCP server integration
pip install "flowforge[artifacts]" # PDF, PPTX, DOCX, charts, images
pip install "flowforge[docs]" # Local MkDocs documentation
pip install "flowforge[all]" # All runtime extrasRequires Python 3.11+
Install directly from GitHub when testing unreleased changes:
pip install "flowforge @ git+https://github.com/JY-1019/FlowForge.git"For local development:
git clone https://github.com/JY-1019/FlowForge.git
cd FlowForge
python -m pip install -e ".[dev,all]"from flowforge import global_config, flow, task, step, FlowForge
from flowforge import LLMConfig, BranchCondition
@global_config(
prompt="Multi-language research assistant",
llm_config=LLMConfig(model="claude-sonnet-4-6", temperature=0.3),
)
class ResearchAgent:
@flow(name="research", prompt="Analyze → Search → Answer")
class ResearchFlow:
@task(name="analyze", prompt="Classify user intent")
class AnalyzeTask:
@step(order=1, prompt="Extract intent and keywords")
async def classify(ctx):
return {"intent": "search", "keywords": ["AI"], "source": "web"}
@task(name="search", prompt="Execute search based on analysis")
class SearchTask:
@step(
order=1,
prompt="Route to the best source",
condition=BranchCondition(field="source", enum=["web", "db"]),
branches={"web": web_handler, "db": db_handler},
fallback=web_handler,
)
async def route(ctx): ...engine = FlowForge.compile(ResearchAgent)
result = await engine.run({"query": "latest AI trends"})print(engine.mermaid())graph TD
global["ResearchAgent"]
global --> research["research"]
research --> analyze["analyze"]
analyze --> classify["classify[1]"]
research --> search["search"]
search --> route["route[1] (branch)"]
Everything is a decorator. No manual graph wiring.
| Decorator | Purpose |
|---|---|
@global_config |
Agent-level settings, LLM config, global tools |
@flow |
Top-level execution unit (nestable) |
@task |
Work unit within a flow (nestable) |
@step |
Atomic action within a task (ordered) |
Flows contain flows. Tasks contain tasks. Build any depth of hierarchy.
@flow(name="pipeline", prompt="Main pipeline")
class Pipeline:
@flow(name="ingestion", prompt="Data ingestion sub-flow")
class Ingestion:
@task(name="fetch", prompt="Fetch data")
class Fetch: ...
@task(name="transform", prompt="Final transformation")
class Transform: ...Conditional routing is built into @step, @task, and @flow:
@step(
order=2,
prompt="Route by document format",
condition=BranchCondition(field="format", enum=["pdf", "csv", "json"]),
branches={"pdf": pdf_handler, "csv": csv_handler, "json": json_handler},
fallback=pdf_handler,
)
async def route_format(ctx): ...Run only specific parts of your agent:
# Run a single flow
result = await engine.run(data, route="search")
# Run a specific task within a flow
result = await engine.run(data, route="research.analyze")
# Run multiple routes
result = await engine.run(data, route=["analysis", "report"])Re-run a task until a quality threshold is met:
@task(
name="quality_check",
prompt="Keep refining until quality is high",
max_loops=5,
loop_condition=lambda out: out.get("score", 0) >= 0.8,
)
class QualityCheck: ...Tools cascade through the hierarchy. Call the LLM from any step:
from flowforge.types import AgentSkill, ClaudeSkill, FunctionTool, MCPServer
@global_config(
prompt="...",
tools=[
MCPServer("https://api.example.com/mcp"),
ClaudeSkill(name="pptx"),
AgentSkill(path=".agents/skills/code-review"),
],
)
class Agent:
@flow(name="f", prompt="...", tools=["calculator"])
class F:
@task(name="t", prompt="...")
class T:
@step(order=1, prompt="Use the calculator tool to solve the query",
tools=["calculator", "pptx", "code-review"])
async def solve(ctx):
result = await ctx.call_llm("Solve: {query} <calculator>")
deck = await ctx.call_llm("Turn the result into slides. <pptx>")
review = await ctx.call_llm("Review the solution. <code-review>")
return resultChild annotations can list string tool references such as
tools=["calculator"]; FlowForge resolves them against globally registered
tool configs at runtime. This is useful for dynamic generated flows because
they can declare intended tools without recreating FunctionTool objects.
FlowForge supports five tool families:
| Tool | Use for | Runtime behavior |
|---|---|---|
FunctionTool |
Local Python functions | FlowForge executes it in the tool-use loop |
HTTPTool |
HTTP APIs | FlowForge calls the endpoint with httpx |
MCPServer |
Remote MCP tools | FlowForge fetches MCP schemas and calls tools/call |
ClaudeSkill |
Anthropic native Skills such as pptx |
Sent to Claude as container.skills |
AgentSkill |
Local standard SKILL.md folders |
Loaded into the model context with <skill-name> |
Use ClaudeSkill when you want Anthropic's native Skills API. Use
AgentSkill when a user has a local Agent Skills folder and you want the same
tools=[...] + <skill-name> workflow across Anthropic, OpenAI, and Google
providers.
Nodes with the same order value run in parallel:
@task(name="t", prompt="...")
class T:
@step(order=1, prompt="Step A") # runs in parallel with Step B
async def a(ctx): ...
@step(order=1, prompt="Step B") # same order = parallel
async def b(ctx): ...
@step(order=2, prompt="Step C") # runs after A and B complete
async def c(ctx): ...Let the agent create missing flows when the compiled DAG does not already cover a user request:
from flowforge import global_config, FlowForge, DynamicRunOptions
@global_config(
prompt="General-purpose AI agent",
llm_config=LLMConfig.for_claude(),
dynamic_flow=True, # enable dynamic generation
)
class MyAgent:
pass # zero static flows is OK
options = DynamicRunOptions(
project_root=".",
persist_generated=True, # save generated code
auto_load_generated=True, # reuse it on the next compile
include_builtin_tools=True, # built-ins for PPT, CSV, charts, etc.
)
engine = FlowForge.compile(MyAgent, dynamic_options=options)
result = await engine.run(
"Make a table of the five tallest mountains in the world",
planning_mode="autonomous",
)Key features:
- Manifest caching — generated flows are saved, loaded on compile, and skipped when already present
- Compact codegen context — only relevant tools and project context are sent to the code generator by default
- 14+ builtin tools — web, files, PPT Master-backed editable PPT, CSV, DOCX, charts, PDF, markdown
- Dynamic MCP setup — generated flows can start declared MCP servers and register their tools for later
<tool>use - Auto artifact detection — "make a PPT" → automatically includes
pptx_create - AST safety validation — blocks dangerous imports/calls before execution
- Contract-first chaining — generated flow output matches downstream input schema
Catch errors before runtime:
- Order uniqueness within tasks
- I/O schema compatibility between steps
- Branch handler output type consistency
- DAG cycle detection
flowforge validate ./agent.py # Check for compile errors
flowforge viz ./agent.py --mermaid # Print Mermaid diagram
flowforge run ./agent.py --query "test" # Execute the agent
flowforge doc-generate ./agent.py # Auto-generate node docs via LLM| File | Shows |
|---|---|
examples/dynamic_bare_agent.py |
Zero-flow agent that generates missing flows at runtime |
examples/dynamic_docx_report_agent.py |
Zero-flow, zero-tool agent that dynamically generates a flow which uses the built-in docx_create tool to materialise a .docx report under ~/test/docx_reports |
examples/dynamic_skill_mcp_agent.py |
Zero-flow agent showing Agent Skill guidance, optional Claude pptx Skill use, compact dynamic codegen, and Playwright/Figma MCP server registration |
examples/dynamic_paper_report_agent.py |
Static pipeline plus dynamic upstream paper search/reporting |
examples/claude_skill_custom_text_agent.py |
Custom Claude Skill proof that prints a marker immediately |
examples/claude_skill_pptx_agent.py |
Anthropic pptx Skill, file ID extraction, and Files API download |
examples/playwright_agent.py |
MCP browser tool integration |
examples/enterprise_agent.py |
Larger nested flow/task composition |
@step → @task → @flow → @global_config → FlowForge.compile()
│ │ │ │ │
│ │ │ │ ▼
│ │ │ │ Build DAG (networkx)
│ │ │ │ Topological sort
│ │ │ ▼ Cycle detection
│ │ │ Scan for flows Validate I/O chains
│ │ ▼ │
│ │ Scan for tasks/flows │
│ ▼ ▼
│ Scan for steps/child tasks ExecutionEngine
▼ (async runner)
Attach metadata to function
- Decorate — Each decorator attaches metadata to the class/function
- Compile —
FlowForge.compile()reads the metadata tree and builds a DAG - Validate — Schema compatibility, order conflicts, and cycles are checked
- Execute — The async engine traverses the DAG, passing data between nodes
flowforge/
├── annotations/ # Decorators, metadata, validators
├── schema/ # DAG compiler, registry, resolver
├── execution/ # Async runners, context, LLM integration
├── dynamic/ # Dynamic flow generation, manifest, meta-flow
├── tools/ # MCP, HTTP, function tool adapters, builtin tools
├── planner/ # AI-driven path selection
├── viz/ # Mermaid & Graphviz rendering
└── cli/ # Typer-based CLI
Full documentation is available via MkDocs Material.
After installing with the [docs] extra, a single command spins up the local docs server
and opens it in your browser — no repo clone required:
pip install "flowforge[docs] @ git+https://github.com/JY-1019/FlowForge.git"
flowforge docsOther useful forms:
flowforge docs --online # just open the published GitHub Pages site
flowforge docs --port 9000 # bind a different port
flowforge docs --build # build a static site into ./site/
flowforge docs --no-open # serve without auto-opening the browserUnder the hood, mkdocs.yml and the full _docs/ tree are shipped as package data
inside the flowforge package, so they're available the moment pip install finishes.
The CLI resolves the bundled mkdocs.yml via importlib-style package path lookup, so it
works identically from a wheel install, an editable install (pip install -e .), or a
fresh git clone.
| Section | Description |
|---|---|
| Getting Started | Installation, hello world, first run |
| Concepts | Architecture, annotations, data flow, DAG compilation |
| Guides | First agent, branch dispatching, nested flows, tools & LLM, visualization, dynamic flow generation |
| API Reference | Decorators, types, engine, errors, CLI |
You can also browse the docs directly on GitHub: flowforge/_docs/
MIT License. See LICENSE for details.
Contributions are welcome. See CONTRIBUTING.md for the development setup, test commands, and packaging checks. Please report security issues privately using the guidance in SECURITY.md.
Jongyeon Keum