Skip to content

PyFlue

PyFlue is a Python-first agent harness framework for building autonomous agents with Markdown skills, persistent sessions, sandboxed filesystem access, shell execution, typed Pydantic outputs, and pluggable harness backends. It adapts the agent harness pattern for Python teams. 

uv add pyflue
pyflue init my-agent
cd my-agent
pyflue run --prompt "Review this project"

pip install pyflue
pyflue init my-agent
cd my-agent
pyflue run --prompt "Review this project"

Why PyFlue

Warning: PyFlue is under active development. The API may change. Pin your dependencies and review changelogs before updating.

PyFlue gives Python developers a framework-shaped agent runtime instead of a collection of low-level primitives. The default backend is Pydantic AI, a typed and model agnostic loop with no LangChain dependency. DeepAgents is available as an optional extra for LangChain users. A stable PyFlue API is layered above the harness.

PyFlue provides Markdown skills, stateful sessions, sandboxed tools, typed outputs, and deployable agent entrypoints for Python teams with Pydantic, Python packaging, and Python-friendly deployment targets.

PyFlue has two boundaries for model driven work, matching the Flue model. A persistent agent keeps sessions over time; a finite workflow runs one bounded operation and returns a result.

# A persistent agent in src/agents/assistant.py
from pyflue import create_agent

default = create_agent(lambda ctx: {"model": "openai:gpt-5.5"})

# A finite workflow in src/workflows/summarize.py
from pyflue import FlueContext, create_agent

agent = create_agent(lambda ctx: {"model": "openai:gpt-5.5"})


async def run(ctx: FlueContext) -> dict:
    harness = await ctx.init(agent)
    session = await harness.session()
    response = await session.prompt(ctx.payload["text"])
    return {"summary": response.text}

See Agents vs Workflows for when to use each.

PyFlue is designed for Python teams that want the ergonomics of a modern agent harness while keeping access to the Python ecosystem. It is useful for coding agents, data workflows, support automation, and service agents that need structured outputs and controlled access to files or shell commands.

Markdown Skills

Define reusable workflows in .agents/skills/*.md with YAML frontmatter.

Stateful Sessions

Persist conversation history with SQLite-backed sessions.

Virtual Sandbox

Read, write, edit, grep, glob, and run shell commands behind policy gates.

Typed Outputs

Validate final results with Pydantic v2.

What You Can Build

PyFlue gives you the core pieces needed for agentic workflows:

  • project instructions from AGENTS.md and CLAUDE.md
  • reusable Markdown skills in .agents/skills
  • stateful sessions backed by SQLite
  • a virtual sandbox with read, write, edit, grep, glob, and shell tools
  • Pydantic validation for typed results
  • cancellation for active prompt, stream, task, and shell operations
  • structured command tools with PyFlueCommand
  • a Python client for deployed PyFlue servers
  • a default Pydantic AI runtime backend, with DeepAgents available as an optional extra
  • a backend registry for OpenAI Agents, Google ADK, and custom harness backends
  • streaming events through Python, CLI, and SSE
  • route triggers for file-based webhook agents
  • chat integration through verified webhooks, dispatch(), and explicit reply tools
  • provider-qualified model selection and provider endpoint overrides
  • secret grants for shell and prompt calls
  • command allowlists and compound-command protection
  • deployment files for Docker/FastAPI, GitHub Actions, GitLab CI, Railway, Render, Fly.io, Vercel, Netlify, and Cloudflare Containers starter projects

Minimal Python Example

from pydantic import BaseModel
from pyflue import init


class FixResult(BaseModel):
    fix_applied: bool
    summary: str


async def main():
    agent = await init(
        model="openai:gpt-5.5",
        harness="deepagents",
        sandbox="virtual",
        allow_write=True,
        allow_shell=True,
    )

    session = await agent.session("fix-123")
    result = await session.skill(
        "triage",
        args={"issue_number": 123},
        result=FixResult,
    )

    if result.fix_applied:
        await session.shell("git status --short")

Next Steps