0.6.1

FastroAI 0.6.1 Release Notes

FastroAI 0.6.1 raises the genai-prices floor to >=0.0.71, so CostCalculator prices the current
generation of models out of the box - notably DeepSeek's V4 family (deepseek-v4-flash,
deepseek-v4-pro), which the older pinned data did not know and therefore priced at zero. A cost of
zero for an unrecognized model is silent, so a stale price floor reads as "free" rather than "unknown"

• raising the floor closes that gap for every consumer on their next lock.

Breaking Changes: None. Dependency floor bump only; all APIs are unchanged.

---

FastroAI 0.6.0 Release Notes

FastroAI 0.6.0 adds dispatch hooks to FastroAgent — async callbacks that fire before and after each agent.run() attempt inside the retry loop. These are the integration points downstream code needs to plug in circuit breakers, rate limiters, and kill switches without monkey-patching the dispatch path. A new DispatchSkippedError exception lets on_before_dispatch short-circuit a dispatch entirely (no retry, no after-hook), and an ErrorCategory enum is provided for callers that want a common vocabulary when classifying outcomes inside on_after_dispatch.

Summary

New:

• FastroAgent(on_before_dispatch=..., on_after_dispatch=...). Per-attempt async hooks wired into StepContext._execute_with_config. Fire on every retry, not just the final outcome. Use them for breakers, rate limiters, and any guard that needs to see each individual try.
• DispatchSkippedError — subclass of FastroAIError. Raise from on_before_dispatch to short-circuit. The retry loop propagates without retrying, and on_after_dispatch is not called. Use for breakers-open / kill-switches / "we already know this will fail."
• ErrorCategory StrEnum (TRANSIENT, PERMANENT, RESOURCE_EXHAUSTION, UNKNOWN) — provided for callers that want to categorize exceptions inside on_after_dispatch. The library doesn't auto-categorize; you map exceptions yourself.
• AgentConfig.timeout_name — optional human-readable label for the configured timeout, surfaced in span attributes and TimeoutError messages emitted by the retry loop. Observability tooling (e.g. Logfire) uses it to distinguish which configured timeout fired without parsing call-site context.

Breaking Changes: None. All additions are opt-in; existing code continues to work unchanged.

Why This Exists

A 50-minute production hang on 2026-05-06 in a downstream worker exposed the gap between "retries are bounded by max_retries" and "downstream code can react to per-attempt outcomes." The 0.5.0 timeout fix bounded individual calls, but circuit-breaker integrations needed something else: visibility into each attempt as it happens, plus a way to fail fast when a breaker has already tripped. Patching that into application code on top of agent.run() worked for one consumer but didn't compose — every caller would have wrapped the same retry loop differently. So the hooks live in the library, fire per attempt, and have well-defined semantics around skipping (DispatchSkippedError) and pre-flight rejection (CostBudgetExceededError).

How It Works

Hooks fire inside StepContext._execute_with_config, the retry loop that wraps agent.run(). Per attempt:

1. on_before_dispatch() if set. Any exception (including DispatchSkippedError) propagates immediately — no retry, no after-hook.
2. agent.run(...) (with timeout if configured).
3. on_after_dispatch(exc_or_none) if set. After-hook errors propagate without retry.
4. On agent success: return. On agent failure: continue retry loop unless the failure was DispatchSkippedError.

CostBudgetExceededError is a pre-flight rejection — it raises in ctx.run() before the dispatch loop is entered, so neither hook fires for it.

Example: Circuit Breaker Integration

from fastroai import FastroAgent, DispatchSkippedError, ErrorCategory

class BreakerOpenError(DispatchSkippedError):
    pass

async def before():
    if breaker.is_open():
        raise BreakerOpenError("downstream breaker open")

async def after(exc: Exception | None):
    if exc is None:
        breaker.record_success()
        return
    category = categorize(exc)  # your app's classifier, returning ErrorCategory
    if category == ErrorCategory.TRANSIENT:
        breaker.record_failure()
    # PERMANENT failures don't count toward the breaker

agent = FastroAgent(
    model="openai:gpt-4o",
    timeout=30,
    timeout_name="summarize_step.dispatch",
    on_before_dispatch=before,
    on_after_dispatch=after,
)

When the breaker is open, before() raises BreakerOpenError (a DispatchSkippedError subclass) — the agent never dispatches, no retry happens, and the application sees the breaker-specific exception type bubble up. When the breaker is closed but a request fails, after() decides whether to count the failure based on the application's categorization. The timeout_name shows up in span attributes so you can tell which configured timeout fired without scanning call sites.

Upgrade Guide

No code changes required. FastroAgent(...) calls without the new params work exactly as before. To opt in:

# Before (still works):
agent = FastroAgent(model="openai:gpt-4o", timeout=30)

# After (with hooks):
agent = FastroAgent(
    model="openai:gpt-4o",
    timeout=30,
    timeout_name="my_step.dispatch",  # optional, observability only
    on_before_dispatch=my_before_hook,
    on_after_dispatch=my_after_hook,
)

Hooks live on the FastroAgent instance, not in AgentConfig — they're Callables and don't serialize. timeout_name is a string so it lives on AgentConfig like the rest of the configuration.

---

What's Changed

• Bump genai-prices floor to 0.0.71 for current model pricing; by @igorbenav in #12

Full Changelog: v0.6.0...v0.6.1

0.6.0

FastroAI 0.6.0 Release Notes

FastroAI 0.6.0 adds dispatch hooks to FastroAgent — async callbacks that fire before and after each agent.run() attempt inside the retry loop. These are the integration points downstream code needs to plug in circuit breakers, rate limiters, and kill switches without monkey-patching the dispatch path. A new DispatchSkippedError exception lets on_before_dispatch short-circuit a dispatch entirely (no retry, no after-hook), and an ErrorCategory enum is provided for callers that want a common vocabulary when classifying outcomes inside on_after_dispatch.

Summary

New:

• FastroAgent(on_before_dispatch=..., on_after_dispatch=...). Per-attempt async hooks wired into StepContext._execute_with_config. Fire on every retry, not just the final outcome. Use them for breakers, rate limiters, and any guard that needs to see each individual try.
• DispatchSkippedError — subclass of FastroAIError. Raise from on_before_dispatch to short-circuit. The retry loop propagates without retrying, and on_after_dispatch is not called. Use for breakers-open / kill-switches / "we already know this will fail."
• ErrorCategory StrEnum (TRANSIENT, PERMANENT, RESOURCE_EXHAUSTION, UNKNOWN) — provided for callers that want to categorize exceptions inside on_after_dispatch. The library doesn't auto-categorize; you map exceptions yourself.
• AgentConfig.timeout_name — optional human-readable label for the configured timeout, surfaced in span attributes and TimeoutError messages emitted by the retry loop. Observability tooling (e.g. Logfire) uses it to distinguish which configured timeout fired without parsing call-site context.

Breaking Changes: None. All additions are opt-in; existing code continues to work unchanged.

Why This Exists

A 50-minute production hang on 2026-05-06 in a downstream worker exposed the gap between "retries are bounded by max_retries" and "downstream code can react to per-attempt outcomes." The 0.5.0 timeout fix bounded individual calls, but circuit-breaker integrations needed something else: visibility into each attempt as it happens, plus a way to fail fast when a breaker has already tripped. Patching that into application code on top of agent.run() worked for one consumer but didn't compose — every caller would have wrapped the same retry loop differently. So the hooks live in the library, fire per attempt, and have well-defined semantics around skipping (DispatchSkippedError) and pre-flight rejection (CostBudgetExceededError).

How It Works

Hooks fire inside StepContext._execute_with_config, the retry loop that wraps agent.run(). Per attempt:

1. on_before_dispatch() if set. Any exception (including DispatchSkippedError) propagates immediately — no retry, no after-hook.
2. agent.run(...) (with timeout if configured).
3. on_after_dispatch(exc_or_none) if set. After-hook errors propagate without retry.
4. On agent success: return. On agent failure: continue retry loop unless the failure was DispatchSkippedError.

CostBudgetExceededError is a pre-flight rejection — it raises in ctx.run() before the dispatch loop is entered, so neither hook fires for it.

Example: Circuit Breaker Integration

from fastroai import FastroAgent, DispatchSkippedError, ErrorCategory

class BreakerOpenError(DispatchSkippedError):
    pass

async def before():
    if breaker.is_open():
        raise BreakerOpenError("downstream breaker open")

async def after(exc: Exception | None):
    if exc is None:
        breaker.record_success()
        return
    category = categorize(exc)  # your app's classifier, returning ErrorCategory
    if category == ErrorCategory.TRANSIENT:
        breaker.record_failure()
    # PERMANENT failures don't count toward the breaker

agent = FastroAgent(
    model="openai:gpt-4o",
    timeout=30,
    timeout_name="summarize_step.dispatch",
    on_before_dispatch=before,
    on_after_dispatch=after,
)

When the breaker is open, before() raises BreakerOpenError (a DispatchSkippedError subclass) — the agent never dispatches, no retry happens, and the application sees the breaker-specific exception type bubble up. When the breaker is closed but a request fails, after() decides whether to count the failure based on the application's categorization. The timeout_name shows up in span attributes so you can tell which configured timeout fired without scanning call sites.

Upgrade Guide

No code changes required. FastroAgent(...) calls without the new params work exactly as before. To opt in:

# Before (still works):
agent = FastroAgent(model="openai:gpt-4o", timeout=30)

# After (with hooks):
agent = FastroAgent(
    model="openai:gpt-4o",
    timeout=30,
    timeout_name="my_step.dispatch",  # optional, observability only
    on_before_dispatch=my_before_hook,
    on_after_dispatch=my_after_hook,
)

Hooks live on the FastroAgent instance, not in AgentConfig — they're Callables and don't serialize. timeout_name is a string so it lives on AgentConfig like the rest of the configuration.

---

Full Changelog: v0.5.0...v0.6.0

0.5.0

FastroAI 0.5.0 Release Notes

FastroAI 0.5.0 fixes a long-standing bug where the timeout parameter on FastroAgent was accepted but never actually enforced — the configured value was stored on AgentConfig and silently dropped at the boundary to pydantic-ai's ModelSettings. Per-request timeouts now reach the underlying model client. Breaking change: the field has been renamed from timeout_seconds to timeout and defaults to None (opt-in) instead of an always-set 120.

Summary

Bug Fix:

• Timeout forwarding works. FastroAgent(timeout=N) now ends up at ModelSettings.timeout=N and is honored by the underlying model client (e.g. OpenAI SDK passes it to chat.completions.create(timeout=N)). Previously the kwarg was stored on AgentConfig but never read from _execute(), so a single LLM call could run for as long as the underlying httpx client's read timeout permitted (typically 600s × max_retries+1 attempts).

Breaking Changes:

• AgentConfig.timeout_seconds → AgentConfig.timeout. The field is now named timeout to match pydantic-ai's ModelSettings.timeout convention and to disambiguate from "max retries" semantics. Code using the old name will fail with a Pydantic validation error.
• Default is now None (opt-in), previously 120. With the default, no per-request timeout is forwarded to the model client and the client's own defaults apply (e.g. OpenAI SDK's 600s read timeout). Code that relied on the silent 120s default — though no such code worked correctly because the default was never enforced — should now pass timeout=120 explicitly.
• DEFAULT_TIMEOUT_SECONDS constant removed from fastroai.agent exports. There is no replacement constant; choose a value at the call site.

The Problem We Solved

Previously, FastroAgent(timeout=N) accepted the kwarg but it never reached the model client. Walked through the call path:

1. FastroAgent.__init__(**kwargs) forwarded kwargs to AgentConfig(**kwargs).
2. Consumers passed timeout=N. Pydantic v2's default extra='ignore' silently dropped the kwarg because AgentConfig had timeout_seconds, not timeout. The constant didn't even reach the config.
3. Even when consumers correctly used timeout_seconds=N, _execute() only forwarded max_tokens and temperature to ModelSettings. The timeout_seconds field was stored but never read. The configured value never reached the model client.

Net result: every consumer with a *_TIMEOUT_SECONDS constant was running with the underlying httpx default, which on OpenAI's SDK is read=600s with max_retries=2 — worst case 30 minutes per LLM call before any error surfaces. This contributed to a 50-minute production hang in a downstream worker on 2026-05-06: a hung pydantic-ai agent call that should have hit a 300s timeout per the configured constant ran for ~30 min until the worker was manually restarted.

The Fix

FastroAgent._build_default_model_settings() (new helper used by both _execute and _execute_stream) reads self.config.timeout and forwards it into ModelSettings.timeout when set. When None, the key is omitted so the model client's defaults apply. The field was renamed timeout_seconds → timeout so the kwarg name matches what consumers actually pass and so it lines up with pydantic-ai's existing ModelSettings.timeout field.

Per-call enforcement now flows: FastroAgent(timeout=300).run(...) → _build_default_model_settings() → ModelSettings(timeout=300) → Agent.run(model_settings=...) → OpenAIChatModel → AsyncOpenAI.chat.completions.create(timeout=300) → httpx (per-attempt deadline). With max_retries=2 on the SDK, the worst-case wall time is (retries + 1) × timeout.

Upgrade Guide

1. Rename timeout_seconds → timeout in your code.

# Before:
agent = FastroAgent(model="openai:gpt-4o", timeout_seconds=60)
config = AgentConfig(timeout_seconds=60)

# After:
agent = FastroAgent(model="openai:gpt-4o", timeout=60)
config = AgentConfig(timeout=60)

2. Decide whether to opt in to the new behavior.

If you don't pass timeout, the model client's own defaults apply (typically generous: ~600s read on OpenAI). For long-running calls this is what you want. For latency-sensitive paths, set timeout explicitly — and now it'll actually be enforced.

3. Drop DEFAULT_TIMEOUT_SECONDS imports.

# Before:
from fastroai.agent import DEFAULT_TIMEOUT_SECONDS

# After: (no replacement; choose a value at the call site)
TIMEOUT = 60

4. Re-validate any constants you previously assumed were enforced.

If you have a *_TIMEOUT_SECONDS constant that you've been passing into FastroAgent and trusting to bound LLM call wall time, that bound was not actually applying. Check whether your value is still the right one — you may have been over-budgeting because the timeouts were getting absorbed by retries that no longer happen as silently.

---

Full Changelog: v0.4.1...v0.5.0

0.4.1

FastroAI 0.4.1 Release Notes

FastroAI 0.4.1 fixes incorrect model tracking when using PydanticAI's FallbackModel and other model wrappers. Cost calculations are now accurate regardless of which model in a fallback chain processes the request.

Summary

Bug Fixes:

• FallbackModel Support: Correctly tracks the actual model that processed the request, not the configured default
• No False Model Assumptions: When model can't be detected (e.g., escape hatch without explicit model), returns model=None and cost=0 instead of assuming gpt-4o

No Breaking Changes - ChatResponse.model is now str | None but existing code checking the model string will continue to work.

The Problem We Solved

Previously, FastroAgent attempted to get the model name from usage.model, but PydanticAI's RunUsage class doesn't have this field. This caused two issues:

1. FallbackModel Misreporting: When using FallbackModel with DeepSeek primary and GPT-4o fallback, the tracked model was always openai:gpt-4o (the config default) regardless of which model actually responded.

2. Incorrect Cost Calculation: Costs were calculated using the wrong model's pricing, potentially overcharging or undercharging by significant amounts.

Example: FallbackModel with DeepSeek

from pydantic_ai import Agent
from pydantic_ai.models.fallback import FallbackModel
from pydantic_ai.models.openai import OpenAIChatModel
from fastroai import FastroAgent

fallback_model = FallbackModel(
    OpenAIChatModel("deepseek-chat"),  # Primary
    OpenAIChatModel("gpt-4o-mini"),    # Fallback
)

pydantic_agent = Agent(model=fallback_model, output_type=str)
agent = FastroAgent(agent=pydantic_agent)

response = await agent.run("Hello")

Metric	Before (v0.4.0)	After (v0.4.1)
response.model	"openai:gpt-4o"	"deepseek-chat"
Cost Basis	GPT-4o pricing	DeepSeek pricing

How It Works

FastroAgent now extracts the model name from ModelResponse.model_name in the message history, which PydanticAI populates with the actual model that processed each request:

# PydanticAI's response structure
response.all_messages()
# Returns:
# [
#     ModelRequest(...),
#     ModelResponse(
#         parts=[TextPart(content='...')],
#         model_name='deepseek-chat',  # <-- Actual model used
#         timestamp=datetime.datetime(...),
#     ),
# ]

Fallback Behavior

When model extraction fails:

1. With explicit model configured: Falls back to the configured model
2. Escape hatch without model: Returns model=None and cost_microcents=0, logs a warning

# Escape hatch without explicit model - model detection required
agent = FastroAgent(agent=custom_pydantic_agent)  # No model=
# If detection fails: model=None, cost=0, warning logged

# Escape hatch with explicit model - has fallback
agent = FastroAgent(agent=custom_pydantic_agent, model="gpt-4o-mini")
# If detection fails: model="gpt-4o-mini"

API Changes

ChatResponse.model

The model field is now str | None:

response = await agent.run("Hello")

if response.model:
    print(f"Model: {response.model}")
    print(f"Cost: ${response.cost_dollars:.6f}")
else:
    print("Model unknown - cost not calculated")
    print(f"Tokens used: {response.total_tokens}")

CostCalculator.calculate_cost()

Now accepts model: str | None and returns 0 for None:

calc = CostCalculator()

# Returns 0 for unknown model
cost = calc.calculate_cost(None, input_tokens=100, output_tokens=50)
assert cost == 0

Upgrade Guide

No changes required for most users. If you're using FallbackModel or custom agents:

1. Verify model tracking: Check that response.model now shows the correct model
2. Handle None model: If using escape hatch without explicit model, handle the case where response.model is None

# Before: assumed gpt-4o pricing even with FallbackModel
# After: correct model and pricing

# If you need to handle unknown models:
if response.model is None:
    logger.warning("Model unknown, cost not calculated")

---

What's Changed

• community stuff by @igorbenav in #7
• chapters 5, 6, 7 by @igorbenav in #8
• fix bug, update docs by @igorbenav in #9

Full Changelog: v0.4.0...v0.4.1

0.4.0

FastroAI 0.4.0 Release Notes

FastroAI 0.4.0 adds enhanced cost tracking with full support for prompt caching, tool usage metrics, and provider-specific usage details. Cost calculations are now up to 18% more accurate when prompt caching is enabled.

Summary

What's New for Users:

• Cache Token Tracking: Accurate cost calculation with prompt caching (90% discount for cached tokens)
• Tool Usage Metrics: Track tool_call_count and request_count for agentic behavior monitoring
• Audio Token Support: Track audio tokens for multimodal models
• Provider Details: Access provider-specific data like reasoning tokens via usage_details

No Breaking Changes - All new fields have sensible defaults. Existing code works without modification.

The Problem We Solved

Previously, FastroAI only tracked input_tokens and output_tokens. When prompt caching was enabled (Anthropic, OpenAI), cached tokens were charged at full price in our calculations, even though providers charge 90% less for cached tokens.

Example: Personal Finance Assistant with cached system prompt

Metric	Before (v0.3.0)	After (v0.4.0)
Reported Cost	$0.00300	$0.00246
Accuracy	Overreported by 18%	Accurate

Now FastroAI extracts cache token counts from the provider and applies the correct discounted rate.

New Response Fields

ChatResponse now includes:

response = await agent.run("Hello!")

# Cache tokens (for prompt caching)
print(response.cache_read_tokens)   # Tokens read from cache (90% cheaper)
print(response.cache_write_tokens)  # Tokens written to cache

# Audio tokens (for multimodal)
print(response.input_audio_tokens)
print(response.output_audio_tokens)

# Request/tool metrics
print(response.request_count)       # API requests made (increases with tool use)
print(response.tool_call_count)     # Number of tool invocations

# Provider-specific details
print(response.usage_details)       # e.g., {"reasoning_tokens": 150} for o1

All fields default to 0 or empty, so existing code continues to work.

Enhanced Cost Calculator

CostCalculator.calculate_cost() now accepts cache and audio token parameters:

from fastroai import CostCalculator

calc = CostCalculator()

# Basic usage (unchanged)
cost = calc.calculate_cost("gpt-4o", input_tokens=1000, output_tokens=500)

# With cache tokens (new)
cost = calc.calculate_cost(
    "claude-3-5-sonnet",
    input_tokens=1000,
    output_tokens=500,
    cache_read_tokens=800,  # 800 tokens at 90% discount
)

The signature uses keyword-only arguments after * for backward compatibility - existing calls work unchanged.

Pricing Overrides with Cache Tokens

Custom pricing overrides now support cache token rates:

calc = CostCalculator()

# Override with explicit cache rates
calc.add_pricing_override(
    model="my-cached-model",
    input_per_mtok=3.00,
    output_per_mtok=15.00,
    cache_read_per_mtok=0.30,   # 90% discount
    cache_write_per_mtok=3.75,  # 25% premium
)

# Or use default discounts (90% read, 25% write premium)
calc.add_pricing_override(
    model="volume-discount-model",
    input_per_mtok=2.00,
    output_per_mtok=8.00,
)
cost = calc.calculate_cost(
    "volume-discount-model",
    input_tokens=1000,
    output_tokens=0,
    cache_read_tokens=800,  # Applies default 90% discount
)

Pipeline Usage Tracking

StepUsage and PipelineUsage also include the new fields:

result = await pipeline.execute(input_data, deps)

# Per-step breakdown
for step_id, usage in result.usage.steps.items():
    print(f"{step_id}: {usage.cache_read_tokens} cached tokens")

# Aggregated totals
print(f"Total cached: {result.usage.total_cache_read_tokens}")
print(f"Total tool calls: {result.usage.total_tool_call_count}")

Documentation Updates

• FastroAgent Guide: Updated response fields table with cache tokens, request_count, tool_call_count
• Cost Calculator Guide: Added "Prompt Caching" section explaining cache token tracking
• API Reference: All schemas now use Field(description=...) for better auto-generated docs

Technical Details

Data Flow:

1. Provider (Anthropic/OpenAI) returns cache token counts in API response
2. PydanticAI extracts via genai_prices.extract_usage()
3. FastroAI extracts from PydanticAI's RunUsage
4. CostCalculator passes to genai_prices.calc_price() for accurate pricing

Prompt Caching Requirements:

• Anthropic: Prompts >= 1024 tokens
• OpenAI: Prompts >= 1024 tokens on gpt-4o models

---

What's Changed

• first 5 learn docs by @igorbenav in #3
• some docs fixes by @igorbenav in #4
• ACost tracking improvements by @igorbenav in #5
• update package version by @igorbenav in #6

Full Changelog: v0.3.0...v0.4.0

0.3.0

FastroAI 0.3.0 Release Notes

FastroAI 0.3.0 adds built-in Logfire integration for production-grade observability. You can now trace your AI calls with Pydantic's Logfire platform without writing custom tracer implementations.

Summary

What's New for Users:

• LogfireTracer: Built-in tracer implementation for Pydantic's Logfire observability platform
• Optional Dependency: Install with pip install fastroai[logfire] to enable Logfire support
• Updated Documentation: Tracing guide now covers LogfireTracer usage

No Breaking Changes - This release is fully backwards compatible.

New Features

LogfireTracer

A new built-in tracer that integrates FastroAI with Pydantic Logfire, providing production-grade distributed tracing, metrics, and error tracking.

Installation:

pip install fastroai[logfire]

# Or with uv
uv add "fastroai[logfire]"

Basic Usage:

import logfire
from fastroai import FastroAgent, LogfireTracer

# Configure logfire once at startup
logfire.configure()

tracer = LogfireTracer()
agent = FastroAgent(model="openai:gpt-4o")
response = await agent.run("Hello!", tracer=tracer)

# View traces in the Logfire dashboard

With Pipelines:

from fastroai import Pipeline, LogfireTracer

tracer = LogfireTracer()
result = await pipeline.execute(
    {"document": doc},
    deps=my_deps,
    tracer=tracer,
)

Key Features:

• Implements the Tracer protocol - drop-in replacement for SimpleTracer
• Automatic span creation with _tags=["fastroai"] for easy filtering
• Metric logging via logfire.info() with trace correlation
• Error logging with full exception info via logfire.error()
• Clear ImportError when logfire is not installed

LogfireTracer vs SimpleTracer:

Feature	SimpleTracer	LogfireTracer
Output	Python logging	Logfire platform
Dashboard	No	Yes
Distributed tracing	No	Yes
Production-ready	Development only	Yes
Dependency	None	logfire package

New Exports

The following are now exported from the main fastroai package:

from fastroai import LogfireTracer  # Logfire tracer implementation

Documentation Updates

• Tracing Guide: Added LogfireTracer to built-in tracers section
• API Reference: Added LogfireTracer documentation
• README: Updated features and installation sections

---

Full Changelog: v0.2.0...v0.3.0

0.2.0

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