Metadata-Version: 2.4
Name: agentbeat
Version: 0.1.1
Summary: Production monitoring for AI agents. Heartbeat monitoring, cost tracking, failure detection.
License: MIT
Project-URL: Homepage, https://agentbeat.dev
Project-URL: Documentation, https://agentbeat.dev
Project-URL: Repository, https://github.com/mohitsharmadl/agentbeat-server
Project-URL: Issues, https://github.com/mohitsharmadl/agentbeat-server/issues
Keywords: ai,agents,monitoring,observability,heartbeat,llm,openai,langchain,crewai
Classifier: Development Status :: 4 - Beta
Classifier: Intended Audience :: Developers
Classifier: Topic :: Software Development :: Libraries
Classifier: Topic :: System :: Monitoring
Classifier: Programming Language :: Python :: 3
Classifier: Programming Language :: Python :: 3.8
Classifier: Programming Language :: Python :: 3.9
Classifier: Programming Language :: Python :: 3.10
Classifier: Programming Language :: Python :: 3.11
Classifier: Programming Language :: Python :: 3.12
Requires-Python: >=3.8
Description-Content-Type: text/markdown
Requires-Dist: requests>=2.28

# AgentBeat Python SDK

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

**Website:** [agentbeat.dev](https://agentbeat.dev)

## Install

```bash
pip install agentbeat
```

## Quick Start

```python
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

```python
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

```python
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

```python
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

```python
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

```bash
# 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:

```python
from agentbeat import AgentBeat

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

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

## Handle Failures

```python
# 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

```python
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:

```bash
# 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](https://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
