fastroai 0.2.0

Lightweight AI orchestration built on PydanticAI

FastroAI

Lightweight AI orchestration built on PydanticAI.

Documentation • Discord • GitHub

---

FastroAI wraps PydanticAI with production essentials: cost tracking in microcents, multi-step pipelines, and tools that handle failures gracefully.

Note: FastroAI is experimental. The API may change between versions.

Features

• Cost Tracking: Automatic cost calculation in microcents. No floating-point drift.
• Pipelines: DAG-based workflows with automatic parallelization.
• Safe Tools: Timeout, retry, and graceful error handling for AI tools.
• Tracing: Protocol-based integration with any observability platform.

Installation

pip install fastroai

Or with uv:

uv add fastroai

Quick Start

from fastroai import FastroAgent

agent = FastroAgent(
    model="openai:gpt-4o",
    system_prompt="You are a helpful assistant.",
)

response = await agent.run("What is the capital of France?")

print(response.content)
print(f"Cost: ${response.cost_dollars:.6f}")

Every response includes token counts and cost. No manual tracking required.

Pipelines

Chain multiple AI steps with automatic parallelization:

from fastroai import FastroAgent, Pipeline

extract = FastroAgent(model="openai:gpt-4o-mini", system_prompt="Extract entities.")
classify = FastroAgent(model="openai:gpt-4o-mini", system_prompt="Classify documents.")

pipeline = Pipeline(
    name="processor",
    steps={
        "extract": extract.as_step(lambda ctx: ctx.get_input("text")),
        "classify": classify.as_step(lambda ctx: ctx.get_dependency("extract")),
    },
    dependencies={"classify": ["extract"]},
)

result = await pipeline.execute({"text": "Apple announced..."}, deps=None)
print(f"Total cost: ${result.usage.total_cost_dollars:.6f}")

Safe Tools

Tools that don't crash when external services fail:

from fastroai import safe_tool

@safe_tool(timeout=10, max_retries=2)
async def fetch_weather(location: str) -> str:
    """Get weather for a location."""
    async with httpx.AsyncClient() as client:
        resp = await client.get(f"https://api.weather.com/{location}")
        return resp.text

If the API times out, the AI receives an error message and can respond gracefully.

Documentation

• Quick Start: Install and run your first agent in 2 minutes.
• Guides: Deep dives into agents, pipelines, tools, and tracing.
• API Reference: Complete reference for all classes and functions.

FastroAI Template

Looking for a complete AI SaaS starter? Check out FastroAI Template: authentication, payments, background tasks, and more built on top of this library.

Support

• Questions & Discussion: Discord
• Bugs & Features: GitHub Issues

License

MIT

---

Built by Benav Labs