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).
- 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.
pip install fastroaiWith Logfire tracing:
pip install fastroai[logfire]Or with uv:
uv add fastroai
uv add "fastroai[logfire]" # With Logfire tracingfrom 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.
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}")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.textIf the API times out, the AI receives an error message and can respond gracefully.
- 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.
Looking for a complete AI SaaS starter? Check out FastroAI Template: authentication, payments, background tasks, and more built on top of this library.
- Questions & Discussion: Discord
- Bugs & Features: GitHub Issues
MIT
Built by Benav Labs