Documentation

pip install agentbudget[langchain]   # LangChain / LangGraph
pip install agentbudget[autogen]    # AutoGen

import agentbudget
import openai

agentbudget.init("$5.00")

# Your existing code — no changes needed
client = openai.OpenAI()
response = client.chat.completions.create(
    model="gpt-4o",
    messages=[{"role": "user", "content": "Hello"}]
)

print(agentbudget.spent())      # e.g. 0.0035
print(agentbudget.remaining())  # e.g. 4.9965
print(agentbudget.report())     # Full cost breakdown

agentbudget.teardown()  # Stop tracking, get final report

# All of these work:
AgentBudget(max_spend="$5.00")
AgentBudget(max_spend="5.00")
AgentBudget(max_spend=5.0)
AgentBudget(max_spend=5)

budget = AgentBudget(
    max_spend="$5.00",
    soft_limit=0.9,               # Warn at 90%
    max_repeated_calls=10,        # Trip after 10 repeated calls
    loop_window_seconds=60.0,     # Within a 60-second window
    on_soft_limit=lambda r: print("Warning: 90% budget used"),
    on_hard_limit=lambda r: alert_ops_team(r),
    on_loop_detected=lambda r: print("Loop detected!"),
)

{
    "session_id": "sess_abc123",
    "budget": 5.00,
    "total_spent": 3.42,
    "remaining": 1.58,
    "breakdown": {
        "llm": {"total": 3.12, "calls": 8, "by_model": {"gpt-4o": 2.80}},
        "tools": {"total": 0.30, "calls": 6, "by_tool": {"serp_api": 0.05}}
    },
    "duration_seconds": 34.2,
    "terminated_by": null,
    "events": [...]
}

agentbudget.init("$5.00")
client = openai.OpenAI()

stream = client.chat.completions.create(
    model="gpt-4o",
    messages=[{"role": "user", "content": "Summarize this report"}],
    stream=True,  # include_usage is added automatically for OpenAI
)
for chunk in stream:
    print(chunk.choices[0].delta.content or "", end="")

print(agentbudget.spent())  # cost recorded after the stream

# async for
async for chunk in await client.chat.completions.create(
    stream=True, ...
):
    process(chunk)

# context manager
with client.chat.completions.create(stream=True, ...) as stream:
    for chunk in stream:
        process(chunk)

import agentbudget
from agentbudget import AgentBudget
import openai

budget = AgentBudget(max_spend="$5.00")
with budget.session() as session:
    # Only this instance is tracked
    client = agentbudget.wrap_client(openai.OpenAI(), session)
    response = client.chat.completions.create(...)  # tracked

    other = openai.OpenAI()
    other.chat.completions.create(...)              # NOT tracked

budget = AgentBudget(
    max_spend="$1.00",
    finalization_reserve=0.05,  # hard limit at $0.95, $0.05 reserved for final call
)

with budget.session() as session:
    # ... do work ...

    if session.would_exceed(estimated_final_cost):
        return "Budget nearly exhausted — here is what was completed: ..."

    # Safe to proceed
    response = session.wrap(client.chat.completions.create(...))

from agentbudget import AgentBudget

budget = AgentBudget(max_spend="$5.00")

async with budget.async_session() as session:
    response = await session.wrap_async(
        client.chat.completions.acreate(
            model="gpt-4o",
            messages=[{"role": "user", "content": "Hello"}]
        )
    )

    @session.track_tool(cost=0.01)
    async def async_search(query):
        return await api.search(query)

with budget.session() as parent:
    child = parent.child_session(max_spend=2.0)
    with child:
        child.track("result", cost=1.50, tool_name="sub_task")

    print(parent.spent)      # 1.50
    print(parent.remaining)  # 8.50

budget = AgentBudget(
    max_spend="$5.00",
    webhook_url="https://your-app.com/api/budget-events",
)

budget = AgentBudget(
    max_spend="$5.00",
    on_soft_limit=lambda r: logger.warning(f"90% used: {r}"),
    on_hard_limit=lambda r: alert_ops_team(r),
    on_loop_detected=lambda r: logger.error(f"Loop: {r}"),
)

pip install agentbudget[langchain]

from agentbudget.integrations.langchain import LangChainBudgetCallback

# Use as a context manager so the session is finalized.
with LangChainBudgetCallback(budget="$5.00") as callback:
    agent.invoke(
        {"input": "Research competitors in the CRM space"},
        config={"callbacks": [callback]},
    )

print(callback.get_report())

from agentbudget.integrations.crewai import CrewAIBudgetMiddleware

with CrewAIBudgetMiddleware(budget="$3.00") as middleware:
    result = middleware.track(
        crew.kickoff(),
        cost=0.50,
        tool_name="crew_run"
    )

print(middleware.get_report())

pip install agentbudget[autogen]

from agentbudget.integrations.autogen import BudgetedAssistantAgent, BudgetedUserProxyAgent

assistant = BudgetedAssistantAgent(name="assistant", budget="$5.00")
user = BudgetedUserProxyAgent(name="user", budget="$5.00")

user.initiate_chat(
    assistant,
    message="Research competitors in the CRM space"
)

print(assistant.get_report())

from agentbudget.integrations.autogen import AutoGenBudgetTracker

tracker = AutoGenBudgetTracker(budget="$5.00")
tracker.attach(existing_assistant)

# BudgetExhausted is raised automatically when the limit is hit
print(tracker.get_report())

AgentBudget(
    max_spend: str | float | int,
    soft_limit: float = 0.9,
    max_repeated_calls: int = 10,
    loop_window_seconds: float = 60.0,
    on_soft_limit: Callable = None,
    on_hard_limit: Callable = None,
    on_loop_detected: Callable = None,
    webhook_url: str = None,
    finalization_reserve: float = 0.0,  # fraction of budget reserved for final step
)