ds-agentic-flow 0.38.0

Minimal wrapper over OpenAI Agent SDK with method chaining and ChatKit integration

AF - Agentic Flow Framework

🚀 v0.38 — SDK 0.14 Refresh. AF now targets openai-agents>=0.14.0,<0.15 (was >=0.3.2). Adds af.SandboxAgent for persistent containerized workspaces, drop-in support for the SDK's new session backends (SQLAlchemy / MongoDB / Redis / Dapr / encrypted), and the new RunConfig fields surfaced in 0.14. Existing af.Agent flows are unchanged. → What's new

Examples now default to GPT-5.5 (released 2026-04-23). No code change required — just pass model="gpt-5.5".

Write agent workflows like regular Python code.

async def research_flow(query: str) -> str:
    # Internal thinking - not saved to session
    async with af.phase("Research"):
        findings = await researcher(query).stream()

    # persist=True saves the final response to session
    async with af.phase("Analysis", persist=True):
        return await analyst(findings).stream()

No graphs. No YAML. No state machines. Just Python.

---

✨ Why the agentic flow approach?

OpenAI Agents SDK is powerful, but multi-agent workflows get verbose fast:

Pure SDK (~50 lines)	AF (~15 lines)
# Manual orchestration result1 = await Runner.run( researcher, messages ) research = result1.final_output result2 = await Runner.run( analyst, [{"role": "user", "content": research}] ) # No streaming, no phases, # no session management...	async def my_flow(query: str) -> str: # Internal thinking - not saved to session async with af.phase("Research"): research = await researcher(query).stream() # persist=True saves the final response to session async with af.phase("Analysis", persist=True): return await analyst(research).stream() runner = af.Runner(flow=my_flow) result = await runner(query)

The agentic flow approach gives you:

• Callable agents - agent(prompt).stream() like PyTorch modules
• Workflow phases - Structure and visibility for multi-step processes
• Streaming mode - .stream() for faster first-token via streaming API
• Session injection - Conversation persistence without global state
• ChatKit integration - Production-ready UI with full-text display

---

🆕 New: .snapshot() — Parallel Agents That Actually Know What's Going On

Ever tried running agents in parallel with asyncio.gather()? You probably hit this tradeoff:

	Knows Context	Parallel-Safe
Default	✅	❌ Race conditions!
.isolated()	❌ Amnesia...	✅
.snapshot()	✅	✅ Best of both!

.snapshot() gives each agent a read-only snapshot of the conversation — they see everything that happened before, but their responses don't interfere with each other.

async with af.phase("Parallel Analysis", persist=True):
    # 3 agents, 1 context, 0 race conditions
    sentiment, entities, summary = await asyncio.gather(
        sentiment_agent(data).snapshot(),   # ← reads context
        entity_agent(data).snapshot(),      # ← reads context
        summary_agent(data).snapshot(),     # ← reads context
    )
    # All three see the same conversation history.
    # None of them write to it. No conflicts. Just works.

TL;DR: .isolated() = no memory, .snapshot() = read-only memory, default = full read/write. Pick the right one for your use case. Learn more →

---

🆕 New: Multi-Provider Support

Use any LLM provider through the SDK's built-in multi-provider system:

# OpenAI (default)
openai_agent = af.Agent(name="openai", model="gpt-5.5", instructions="...")

# Anthropic via LiteLLM
claude_agent = af.Agent(name="claude", model="litellm/anthropic/claude-sonnet-4-20250514", instructions="...")

# Runtime model switch
from agents import RunConfig
result = await agent("prompt").run_config(RunConfig(model="gpt-5.5")).stream()

No AF-specific configuration needed — the SDK handles provider routing via model name prefixes. See Multi-Provider Guide.

Security: LiteLLM 1.82.7/1.82.8 were compromised on PyPI (credential exfiltration). Install with pip install "litellm>=1.82.6,!=1.82.7,!=1.82.8" and audit before upgrading.

---

🆕 New: Sandbox Agents — Agents That Live in a Workspace

agents.sandbox.SandboxAgent (introduced in openai-agents 0.14) gives an agent a persistent containerized workspace — files survive across calls, state can be snapshotted and resumed. AF wraps it as af.SandboxAgent, which is an af.Agent subclass: every existing modifier (.stream(), .silent(), .snapshot(), .isolated(), .max_turns(), .context(), .run_config()) just works.

import agentic_flow as af
from agents import RunConfig, SQLiteSession
from agents.sandbox import Manifest, SandboxRunConfig

# Plan agent: thinking only — no sandbox needed
planner = af.Agent(
    name="planner",
    instructions="Read the spec and produce a step-by-step implementation plan.",
    model="gpt-5.5",
)

# Coder agent: persistent workspace declared as part of agent identity
coder = af.SandboxAgent(
    name="coder",
    instructions="Implement the plan in Python.",
    model="gpt-5.5",
    default_manifest=Manifest(version="1", root="/work", entries=[]),
)

# Flow is just business logic — it does not know about execution environment
async def implement_flow(spec: str) -> str:
    async with af.phase("Plan"):
        plan = await planner(spec).stream()
    async with af.phase("Implement", persist=True):
        return await coder(plan).stream()

# Runner injects the execution environment (Session + sandbox transport) via contextvars
runner = af.Runner(
    flow=implement_flow,
    session=SQLiteSession("chat.db"),
    default_run_config=RunConfig(sandbox=SandboxRunConfig(...)),
)

Manifest is the workspace declaration and belongs to the agent (WHAT). SandboxRunConfig is the runtime transport (which sandbox client, which concurrency limits, which snapshot to resume) and belongs to the Runner (WHERE-environment). Per-call overrides remain available via the existing .run_config() modifier — Runner default + ExecutionSpec modifier compose so that the most-specific setting wins. Manifest, SandboxRunConfig, and the memory/snapshot configs come straight from the SDK; AF intentionally doesn't re-wrap them. Learn more →

---

📋 Requirements

• Python 3.12+
• uv (recommended) or pip
• OpenAI API key

---

📦 Installation

With uv (recommended)

# Install uv if you don't have it
curl -LsSf https://astral.sh/uv/install.sh | sh

# Add AF to your project
uv add git+https://github.com/daiichisankyo/AgenticFlow.git

With pip (alternative)

pip install git+https://github.com/daiichisankyo/AgenticFlow.git

Set your API key

Create a .env.local file from the example:

cp .env.example .env.local
# Edit .env.local and add your OpenAI API key

Or set it directly in your environment:

export OPENAI_API_KEY="your-api-key"

---

🚀 Quickstart Code

import agentic_flow as af

# Define agents
researcher = af.Agent(
    name="researcher",
    instructions="Research the topic thoroughly.",
    model="gpt-5.5",
)

writer = af.Agent(
    name="writer",
    instructions="Write clear, engaging content.",
    model="gpt-5.5",
)

# Define workflow as a regular async function
async def blog_flow(topic: str) -> str:
    # Internal thinking - not saved to session
    async with af.phase("Research"):
        research = await researcher(topic).stream()

    # persist=True saves the final response to session
    async with af.phase("Writing", persist=True):
        return await writer(f"Write about: {research}").stream()

# Run
runner = af.Runner(flow=blog_flow)
article = await runner("quantum computing")

---

💡 Core Concepts

AF is built on three primitives:

• Agent - Callable wrapper around SDK Agent. Returns ExecutionSpec for deferred execution.
• SandboxAgent - Subclass of Agent that constructs agents.sandbox.SandboxAgent (persistent workspace, snapshots, sandbox memory). Uses the same ExecutionSpec and modifiers.
• ExecutionSpec - Lazy specification configured with modifiers (.stream(), .isolated(), .snapshot(), .silent(), .max_turns())
• phase - Context manager for workflow boundaries. Controls session persistence with persist=True.

For details, see Concepts.

---

🔄 Patterns

Sequential Pipeline

async def pipeline(input: str) -> str:
    # Internal processing - not saved to session
    async with af.phase("Extract"):
        entities = await extractor(input).stream()

    async with af.phase("Enrich"):
        enriched = await enricher(entities).stream()

    # persist=True saves the final response to session
    async with af.phase("Format", persist=True):
        return await formatter(enriched).stream()

Conditional Branching

async def smart_process(data: str) -> str:
    # Internal classification - not saved to session
    async with af.phase("Classify"):
        category = await classifier(data).stream()

    # persist=True on whichever branch returns the final response
    if "technical" in category:
        async with af.phase("Technical Analysis", persist=True):
            return await tech_analyst(data).stream()
    else:
        async with af.phase("General Summary", persist=True):
            return await summarizer(data).stream()

Iterative Refinement

async def refine(draft: str) -> str:
    for i in range(3):
        # Internal review - not saved to session
        async with af.phase(f"Review #{i+1}"):
            feedback = await critic(draft).stream()

        if "APPROVED" in feedback:
            return draft

        # Internal revision - not saved to session
        async with af.phase(f"Revise #{i+1}"):
            draft = await reviser(f"{draft}\n\nFeedback: {feedback}").stream()

    # Note: This pattern returns the draft directly without session save.
    # Add a final phase with persist=True if you need to save the result.
    return draft

Parallel Execution

import asyncio

async def parallel_analysis(data: str) -> dict:
    async with af.phase("Parallel Analysis", persist=True):
        # snapshot() reads phase context but doesn't write (concurrent-safe)
        sentiment, entities, summary = await asyncio.gather(
            sentiment_agent(data).snapshot(),
            entity_agent(data).snapshot(),
            summary_agent(data).snapshot(),
        )

        return await synthesizer(
            f"Sentiment: {sentiment}\nEntities: {entities}\nSummary: {summary}"
        ).stream()

---

🔌 ChatKit Integration

from pathlib import Path
import agentic_flow as af
from agents import SQLiteSession
from chatkit.server import ChatKitServer
from fastapi import FastAPI, Request
from fastapi.responses import StreamingResponse

# Persist data in a known location
DATA_DIR = Path(__file__).parent / "data"
DATA_DIR.mkdir(exist_ok=True)

app = FastAPI()


class MyServer(ChatKitServer):
    async def respond(self, thread, item, context):
        user_message = item.content[0].text if item else ""
        session = SQLiteSession(
            session_id=thread.id,
            db_path=str(DATA_DIR / "chat.db"),
        )
        runner = af.Runner(flow=my_flow, session=session)
        async for event in af.chatkit.run_with_chatkit_context(
            runner, thread, self.store, context, user_message
        ):
            yield event


server = MyServer(store)


@app.post("/chatkit")
async def chatkit_endpoint(request: Request):
    result = await server.process(await request.body(), {})
    return StreamingResponse(result, media_type="text/event-stream")

Each af.phase() automatically creates workflow boundaries for proper reasoning display.

---

🎮 Try the Demos

First, install sample dependencies:

uv sync --group sample

CLI Quickstart

Experience the core concepts interactively:

uv run --group sample python sample/quickstart.py

Demonstrates:

1. Flow & Runner separation
2. Declaration vs execution (agent(prompt) returns spec, await executes)
3. Modifiers (.stream(), .isolated(), .silent())
4. Typed output with Pydantic

Guide TUI

Interactive Textual UI that answers questions about AF:

uv run --group sample python -m sample.guide.cli

Guide Web Server (FastAPI + ChatKit)

Start the backend API:

uv run --group sample uvicorn sample.guide.server:app --reload --port 8000

Guide Frontend (Next.js + ChatKit)

In a separate terminal:

cd sample/guide/frontend
npm install
npm run dev

Visit http://localhost:3000

---

📚 Documentation

• Concepts - Call-Spec Discipline, ExecutionSpec, Phase, Modifiers
• API Reference - Complete API documentation
• Context Resolution - Advanced context management

For the complete documentation site, visit https://daiichisankyo.github.io/AgenticFlow/

---

Contributing

# Clone and setup
git clone https://github.com/daiichisankyo/AgenticFlow.git
cd AgenticFlow
uv sync --group dev

# Run tests
uv run pytest

# Run linter
uv run ruff check src/

1. Fork the repository
2. Create a feature branch
3. Make your changes
4. Run tests and linter
5. Submit a pull request

---

📄 License

MIT - see LICENSE for details.