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

Note: FastroAI is experimental, it was extracted into a package from code that we had in production in different contexts. We built it for ourselves but you're free to use and contribute. The API may change between versions and you'll probably find bugs, we're here to fix them. Use in production at your own risk (we do).

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: Built-in Logfire integration, or bring your own observability platform.

Installation

pip install fastroai

With Logfire tracing:

pip install fastroai[logfire]

Or with uv:

uv add fastroai
uv add "fastroai[logfire]"  # With Logfire tracing

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.

Ship your SaaS faster with FastroAI →

Support

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

License

MIT

Built by Benav Labs