AgentBeat Python SDK

Production monitoring for AI agents. Know when your agents fail, overspend, or go silent.

Website: agentbeat.dev

Install

pip install agentbeat

Quick Start

from agentbeat import AgentBeat

# Initialize with your agent's slug and token from agentbeat.dev/dashboard
ab = AgentBeat("https://api.agentbeat.dev", "my-agent", "your-agent-token")

# Option 1: Context manager (recommended)
# Automatically marks the run as completed or failed
with ab.run() as ctx:
    result = my_agent_function()
    ctx.items_processed = len(result)
    ctx.add_cost(0.12)
    ctx.model = "gpt-4o"
    ctx.confidence = 0.95

# Option 2: Simple heartbeat (for cron jobs, scripts)
ab.heartbeat()

# Option 3: Manual start/complete
run_id = ab.start()
# ... your code ...
ab.complete(run_id=run_id, items_processed=42, cost_usd=0.05)

Track Steps in Multi-Step Workflows

with ab.run() as ctx:
    with ctx.timed_step("fetch_data"):
        data = fetch_from_api()

    with ctx.timed_step("process", model="gpt-4o"):
        result = llm_process(data)
        ctx.add_cost(0.08)
        ctx.add_tokens(input_tokens=1200, output_tokens=350)

    with ctx.timed_step("save_results"):
        save_to_db(result)
        ctx.items_processed = len(result)

Use with OpenAI / Anthropic

from agentbeat import AgentBeat
from openai import OpenAI

ab = AgentBeat("https://api.agentbeat.dev", "my-openai-agent", "token")
client = OpenAI()

with ab.run() as ctx:
    response = client.chat.completions.create(
        model="gpt-4o",
        messages=[{"role": "user", "content": "Hello"}]
    )
    ctx.model = "gpt-4o"
    ctx.add_tokens(
        input_tokens=response.usage.prompt_tokens,
        output_tokens=response.usage.completion_tokens
    )
    ctx.add_cost(response.usage.prompt_tokens * 0.0025 / 1000
                 + response.usage.completion_tokens * 0.01 / 1000)

Use with LangChain

from agentbeat import AgentBeat
from langchain_openai import ChatOpenAI
from langchain.agents import create_react_agent, AgentExecutor

ab = AgentBeat("https://api.agentbeat.dev", "my-langchain-agent", "token")

with ab.run() as ctx:
    llm = ChatOpenAI(model="gpt-4o")
    agent = create_react_agent(llm, tools, prompt)
    executor = AgentExecutor(agent=agent, tools=tools)
    result = executor.invoke({"input": "research AI trends"})
    ctx.items_processed = 1
    ctx.model = "gpt-4o"

Use with CrewAI

from agentbeat import AgentBeat
from crewai import Crew

ab = AgentBeat("https://api.agentbeat.dev", "my-crew", "token")

with ab.run() as ctx:
    crew = Crew(agents=[...], tasks=[...])
    result = crew.kickoff()
    ctx.items_processed = len(result.tasks_output)
    ctx.model = "gpt-4o"

Monitor a Cron Job or Shell Script

# Add this one line at the end of your script
curl -s https://api.agentbeat.dev/a/my-cron-job/heartbeat \
  -H "X-Agent-Token: your-token" > /dev/null

Or in Python:

from agentbeat import AgentBeat

ab = AgentBeat("https://api.agentbeat.dev", "daily-etl", "token")

# At the end of your script
ab.heartbeat()

Handle Failures

# The context manager automatically reports failures
with ab.run() as ctx:
    raise ValueError("something broke")
# AgentBeat records: status=failed, error_message="something broke"

# Manual failure reporting
run_id = ab.start()
try:
    do_work()
    ab.complete(run_id=run_id, items_processed=100)
except Exception as e:
    ab.fail(run_id=run_id, error_message=str(e))
    raise

Decorator

from agentbeat import AgentBeat, track_run

ab = AgentBeat("https://api.agentbeat.dev", "my-agent", "token")

@track_run(ab)
def my_agent_task(ctx):
    ctx.items_processed = 50
    ctx.add_cost(0.12)
    ctx.model = "gpt-4o"

my_agent_task()  # Automatically tracked

API Reference

AgentBeat(base_url, agent_slug, agent_token)

Create a client for a specific agent.

ab.start(metadata=None) -> str

Start a new run. Returns run_id.

ab.complete(run_id=None, items_processed=None, cost_usd=None, tokens_input=None, tokens_output=None, model=None, confidence=None)

Complete a run with metrics. If run_id is None, completes the latest run.

ab.fail(run_id=None, error_message="", cost_usd=None)

Mark a run as failed.

ab.heartbeat()

Send a simple heartbeat ping. Use for cron jobs and scripts.

ab.step(run_id, name, status="completed", duration_ms=None, cost_usd=None)

Report a step within a run.

ab.run(metadata=None) -> context manager

Context manager that auto-calls start/complete/fail. Yields a RunContext with:

• ctx.items_processed - number of items processed
• ctx.items_failed - number of items that failed
• ctx.cost_usd - total cost in USD
• ctx.model - model name (e.g. "gpt-4o")
• ctx.confidence - confidence score 0.0-1.0
• ctx.add_cost(usd) - accumulate cost
• ctx.add_tokens(input_tokens, output_tokens) - accumulate token usage
• ctx.timed_step(name) - context manager for timed steps

HTTP API

No SDK required. Use any language with HTTP:

# Start a run
curl -X POST https://api.agentbeat.dev/a/{slug}/start \
  -H "X-Agent-Token: {token}"

# Complete a run
curl -X POST https://api.agentbeat.dev/a/{slug}/complete \
  -H "X-Agent-Token: {token}" \
  -H "Content-Type: application/json" \
  -d '{"items_processed": 42, "cost_usd": 0.12, "model": "gpt-4o"}'

# Report failure
curl -X POST https://api.agentbeat.dev/a/{slug}/fail \
  -H "X-Agent-Token: {token}" \
  -H "Content-Type: application/json" \
  -d '{"error_message": "API timeout"}'

# Simple heartbeat
curl https://api.agentbeat.dev/a/{slug}/heartbeat \
  -H "X-Agent-Token: {token}"

What AgentBeat Monitors

• Heartbeat: alerts when your agent stops running
• Cost: tracks LLM spend per agent, per run, with budget limits
• Failures: detects repeated failures (3 of last 5 runs)
• Steps: tracks multi-step workflow progress
• Alerts: Email, Telegram, Slack, Webhook

Get Started

1. Sign up at agentbeat.dev
2. Create an agent in the dashboard
3. pip install agentbeat
4. Add 3 lines to your code
5. Done — your agent is monitored