agentbeat 0.1.1

Production monitoring for AI agents. Heartbeat monitoring, cost tracking, failure detection.

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