Metadata-Version: 2.4
Name: agentsonar
Version: 0.1.3
Summary: Detect coordination failures in multi-agent AI systems (CrewAI, LangGraph, and more)
Author: AgentSonar
License: Apache-2.0
Keywords: agents,ai,crewai,detection,langgraph,monitoring,multi-agent,observability
Classifier: Development Status :: 4 - Beta
Classifier: Intended Audience :: Developers
Classifier: License :: OSI Approved :: Apache Software License
Classifier: Operating System :: OS Independent
Classifier: Programming Language :: Python :: 3
Classifier: Programming Language :: Python :: 3 :: Only
Classifier: Programming Language :: Python :: 3.10
Classifier: Programming Language :: Python :: 3.11
Classifier: Programming Language :: Python :: 3.12
Classifier: Programming Language :: Python :: 3.13
Classifier: Topic :: Software Development :: Libraries :: Python Modules
Classifier: Topic :: System :: Monitoring
Requires-Python: >=3.10
Requires-Dist: networkx>=3.0
Provides-Extra: all
Requires-Dist: crewai>=0.70.0; extra == 'all'
Requires-Dist: langchain-core>=0.3.0; extra == 'all'
Requires-Dist: langgraph>=0.2.0; extra == 'all'
Provides-Extra: crewai
Requires-Dist: crewai>=0.70.0; extra == 'crewai'
Provides-Extra: langgraph
Requires-Dist: langchain-core>=0.3.0; extra == 'langgraph'
Requires-Dist: langgraph>=0.2.0; extra == 'langgraph'
Description-Content-Type: text/markdown

# AgentSonar

AgentSonar detects coordination failures in multi-agent AI systems —
cycles, repetitive delegation, and runaway throughput — before they
burn through your token budget. **Two lines of code to integrate.**

## Install

```bash
pip install agentsonar[crewai]      # for CrewAI
pip install agentsonar[langgraph]   # for LangGraph / LangChain
pip install agentsonar[all]         # both
```

Framework integrations are optional extras — install only what you need.

## Usage

### CrewAI

```python
from agentsonar import AgentSonarListener

sonar = AgentSonarListener()
# ...run your crew normally. Detection happens automatically.
```

→ Full runnable example: [CrewAI minimal setup](#crewai-minimal-setup).

### LangGraph / LangChain

Two equivalent patterns — pick whichever fits your existing code better.

**Pattern 1: `monitor()` wrapper (recommended if you already have callbacks)**

```python
from agentsonar import monitor

graph = monitor(graph)
result = graph.invoke(input)
```

`monitor()` wraps the compiled graph so every `invoke`/`stream`/`ainvoke`/
`astream` call auto-injects AgentSonar's callback into your config — without
overriding any callbacks you pass. If you already have your own callbacks:

```python
graph = monitor(graph)
result = graph.invoke(input, config={"callbacks": [my_cb]})
# Both callbacks run: [my_cb, AgentSonarCallback()]
```

**Pattern 2: direct callback injection**

```python
from agentsonar import AgentSonarCallback

result = graph.invoke(
    input,
    config={"callbacks": [AgentSonarCallback()]},
)
```

Use this pattern when you want explicit control over callback order. You're
responsible for merging AgentSonar with any existing callbacks yourself —
the `monitor()` wrapper handles that automatically.

→ Full runnable example: [LangGraph minimal setup](#langgraph-minimal-setup).

---

That's the whole API. Zero config required to get started — no API keys,
no accounts. Alerts stream to stderr as they fire and land in per-run
log files on disk. See [Output](#output) for everything AgentSonar writes
and [Configuration](#configuration) for the knobs you can tune.

## Output

Every run creates its **own session directory** under `agentsonar_logs/`
in your working directory. All artifacts for a single run live together:

```
your_project/
└── agentsonar_logs/
    ├── .gitignore                                       # auto-written, contains *
    ├── latest                                            # plain text: current run dir name
    ├── run-2026-04-11_07-03-23-ancient-ember/            # most recent run
    │   ├── timeline.jsonl                                # every event, JSONL
    │   ├── alerts.log                                    # human-readable signal-only
    │   ├── report.json                                   # structured summary report
    │   └── report.html                                   # standalone HTML report
    └── run-2026-04-11_07-03-13-quiet-blossom/            # previous run
        └── ...
```

Live alerts also stream to **stderr** as they fire, prefixed with
`[SONAR HH:MM:SS.mmm]`.

### Calling shutdown()

The JSON and HTML reports are generated on `shutdown()` — so **how**
you end your run determines whether those two files get written.

**LangGraph / LangChain** — call `shutdown()` yourself when the run
completes. Both wrapper patterns expose it:

```python
# Pattern 1: monitor() wrapper — shutdown lives on the wrapper
graph = monitor(graph)
graph.invoke(input)
graph.shutdown()   # ← writes report.json + report.html, closes log files
```

```python
# Pattern 2: direct AgentSonarCallback — shutdown lives on the callback
sonar = AgentSonarCallback()
graph.invoke(input, config={"callbacks": [sonar]})
sonar.shutdown()   # ← writes report.json + report.html, closes log files
```

If you forget to call it, `timeline.jsonl` and the stderr stream are
still captured (they flush event-by-event), but the structured
`report.json` / `report.html` summary files won't be generated for that run.

**CrewAI** — no teardown needed. `AgentSonarListener` hooks
`CrewKickoffCompletedEvent` on the CrewAI event bus and runs
`shutdown()` automatically when `crew.kickoff()` finishes. You get the
full output set — including `report.json` and `report.html` — without
any extra code.

### What `report.html` looks like

The standalone HTML report (`report.html`) is a self-contained page —
no external CSS or JavaScript, no network requests, safe to email,
archive, or commit as a debugging artifact. Each coordination event
renders as a card with its severity, failure class (with hover
tooltip), summary, fingerprint, and expandable topology / thresholds
blocks. Dark mode respects your system preference and persists
across runs. Open it with your browser — or forward to a colleague
and they'll see the same interactive view without any install steps.

### Realtime tailing

`timeline.jsonl` is flushed on every event, so you can watch coordination
problems land as they happen — useful when a long-running crew is behaving
oddly and you want to catch the exact delegation that triggered an alert:

```bash
# Unix / macOS — tail the current run's timeline
tail -f "agentsonar_logs/$(cat agentsonar_logs/latest)/timeline.jsonl"

# Only the signal-only (alerts) view
tail -f "agentsonar_logs/$(cat agentsonar_logs/latest)/alerts.log"
```
```powershell
# Windows PowerShell — -Wait is the tail-follow equivalent
Get-Content "agentsonar_logs/$(Get-Content agentsonar_logs/latest)/timeline.jsonl" -Wait
```

Each JSONL line is a self-contained record with `ts`, `level`, `event`,
and a `data` payload — easy to pipe through `jq` or a custom parser if
you want realtime dashboards during a run.

### Clean-run signals

When the run completes without any WARNING or CRITICAL alerts, AgentSonar
tells you explicitly in every channel instead of leaving you to wonder:

- **HTML report** — a green "✓ No coordination failures detected" banner
  replaces the event cards.
- **`timeline.jsonl`** — the final `session_end` record carries
  `"clean_run": true`, `"warning_count": 0`, `"critical_count": 0`, and
  a plain-English `message` field. Downstream parsers can gate on the
  single boolean instead of walking the whole alerts list.
- **stderr** — a green `✓ No coordination failures detected — clean run.`
  line prints right under the summary banner.

Non-clean runs flip `clean_run` to `false` and the `message` includes
the severity breakdown (e.g. `"2 CRITICAL, 5 WARNING coordination alert(s)
detected during this run."`).

### Run naming, latest pointer, retention

Run directories are named `run-<ISO date>_<time>-<adjective>-<noun>`
(e.g. `run-2026-04-11_07-03-23-ancient-ember`). The timestamp sorts
chronologically; the slug is memorable enough to say out loud and is
deterministic from the session id.

`agentsonar_logs/latest` is a plain text file pointing at the newest
run directory name. Open the newest HTML report with:

```bash
# Unix / macOS
open "agentsonar_logs/$(cat agentsonar_logs/latest)/report.html"
```
```powershell
# Windows PowerShell
Invoke-Item "agentsonar_logs/$(Get-Content agentsonar_logs/latest)/report.html"
```

AgentSonar keeps the **20 most recent runs** by default and prunes
older ones on every new session. Configure via `AGENTSONAR_KEEP_RUNS`
or `config={"keep_runs": N}`; set to `0` to disable pruning.

The auto-written `.gitignore` inside `agentsonar_logs/` contains `*`,
so every log and report is git-ignored by default.

### Opt-out

If you don't want the JSON/HTML reports, disable them in the config:

```python
sonar = AgentSonarCallback(config={"auto_export_on_shutdown": False})
```

### Manual export

If you want to export at multiple checkpoints during a run, or want the
full timeline view (`dedupe=False`), call the exporters explicitly:

```python
from agentsonar._output.json_export import export_json
from agentsonar._output.html_report import export_html

events = sonar.engine.get_recent_events()

# Summary view (default): one line per root cause
export_json(events)
export_html(events, title="Checkpoint 1")

# Timeline view: every state transition preserved
export_json(events, dedupe=False)
```


Configuration reference

Every entry point — `monitor()`, `AgentSonarCallback()`, and
`AgentSonarListener()` — accepts an optional `config` dict to override
the defaults. Pass only the keys you want to change; everything else
keeps its default.

```python
graph = monitor(graph, config={
    # Rate limits — raise these when you're running at demo speeds
    # (much faster than any real LLM workload) and don't want the
    # circuit breaker to trip before detection has a chance to see
    # the full pattern.
    "per_edge_limit": 99999,    # default 10    events per edge per window
    "global_limit":   99999,    # default 200   events total per window
    "window_size":    180.0,    # default 180.0 seconds

    # Alert severity thresholds — how many rotations/events before
    # WARNING and CRITICAL fire. Lower them for tight tests, raise
    # them for noisy production workloads.
    "warning_threshold":  5,    # default 5
    "critical_threshold": 15,   # default 15

    # Output location and retention
    "log_dir":        ".",      # parent directory for agentsonar_logs/
    "keep_runs":      20,       # most recent run dirs to keep (0 = no pruning)
    "console_output": True,     # stream colored alerts to stderr
    "file_output":    True,     # write timeline.jsonl + alerts.log
    "auto_export_on_shutdown": True,  # write report.json + report.html
})
```

The same dict works for every entry point:

```python
AgentSonarCallback(config={"per_edge_limit": 99999})
AgentSonarListener(config={"warning_threshold": 3})
```

The rate-limit knobs are the most common override. AgentSonar's default
circuit breaker is tuned for real LLM workloads (a few hundred ms per
call); scripted demos and unit tests fire events orders of magnitude
faster, which trips the breaker before the downstream cycle/repetition
detectors get a chance to fire. Bumping `per_edge_limit` and
`global_limit` to a large value disables that short-circuit for local
runs.

## What gets detected

Every detected event carries a `failure_class` string that names the kind
of problem. The classes AgentSonar currently surfaces:

- **`cyclic_delegation`** — Agents are stuck in a loop. A delegates to B,
  B to C, C back to A. Usually means an exit condition is missing — a
  reviewer that never approves, a planner that always says "revise".

- **`repetitive_delegation`** — One agent keeps calling another without
  making progress. `A → B` fires many times in a short window with no
  `B → A` return. Usually means A can't make a decision without B and
  isn't getting what it needs.

- **`resource_exhaustion`** — The system is processing events faster
  than it can sustain. Either one edge is being hammered or total
  throughput is over budget. Indicates a runaway agent or throttling
  that's too loose.

**Coming soon** (reserved class names; no-op today):
`cascade_failure`, `authority_violation`, `deadlock`, `agent_stall`,
`token_velocity_anomaly`.

The HTML report shows a hover tooltip on every failure class badge with
the same plain-English description.

## Coordination fingerprint

Every detected failure carries a `coordination_fingerprint` like
`sha256:5c102a66e1104c47`. It's a stable ID for the failure pattern:
the same failure (same agents in the same shape) always produces the
same fingerprint, regardless of when it's detected or how many times
it re-escalates. You use it for:

- **Dedup.** `WARNING → CRITICAL` escalations share a fingerprint, so
  the summary view collapses to one row at the highest severity.
- **Root-cause grouping.** When a cycle is firing, the repetitive-edge
  alerts on each cycle edge are suppressed in the summary view.
- **Cross-run correlation.** Grep any log file for the fingerprint to
  find every time the same pattern fired, across sessions.

You never compute fingerprints yourself — they're just stable IDs
you can use to talk about "the same failure" across time.

## Host safety

AgentSonar never crashes your app. If anything inside the SDK fails,
detection degrades silently to a no-op and your crew / graph / API
keeps running. The kill switch is `AGENTSONAR_DISABLED=1` (also accepts
`true`, `yes`, `on`, `enabled`, case-insensitive) — set it in the
environment to disable AgentSonar without editing code. When running
in degraded mode, `get_summary()` returns `{"degraded": True, ...}`
so you can alert on it from your own dashboards.

## Full minimal setup

Copy-pasteable starting points. The AgentSonar lines are marked — the
rest is plain framework code. Both examples are self-contained and
runnable as-is.

### LangGraph minimal setup

```python
# pip install agentsonar[langgraph]
import operator
from typing import Literal
from langchain_core.messages import AIMessage, AnyMessage, HumanMessage
from langgraph.graph import END, START, StateGraph
from typing_extensions import Annotated, TypedDict

from agentsonar import monitor                                    # ← AgentSonar

class State(TypedDict):
    messages: Annotated[list[AnyMessage], operator.add]
    iteration: int

def planner(s):    return {"messages": [AIMessage(content="plan")]}
def researcher(s): return {"messages": [AIMessage(content="research")]}
def reviewer(s):   return {"messages": [AIMessage(content="revise")],
                           "iteration": s.get("iteration", 0) + 1}
def loop(s) -> Literal["planner", "__end__"]:
    return END if s.get("iteration", 0) >= 8 else "planner"

b = StateGraph(State)
for name, fn in [("planner", planner), ("researcher", researcher), ("reviewer", reviewer)]:
    b.add_node(name, fn)
b.add_edge(START, "planner")
b.add_edge("planner", "researcher")
b.add_edge("researcher", "reviewer")
b.add_conditional_edges("reviewer", loop)

graph = monitor(b.compile())                                      # ← AgentSonar
graph.invoke({"messages": [HumanMessage(content="go")], "iteration": 0},
             config={"recursion_limit": 50})
graph.shutdown()                                                  # ← AgentSonar
```


### CrewAI minimal setup

```python
# pip install agentsonar[crewai]
# export OPENAI_API_KEY=sk-...
from crewai import Agent, Crew, Process, Task

from agentsonar import AgentSonarListener                         # ← AgentSonar
sonar = AgentSonarListener()                                      # ← AgentSonar

researcher = Agent(role="Researcher", goal="Gather info on the topic.",
                   backstory="Senior researcher.", allow_delegation=False)
writer     = Agent(role="Writer", goal="Write a short summary.",
                   backstory="Technical writer.", allow_delegation=False)
manager    = Agent(role="Manager", goal="Coordinate researcher and writer.",
                   backstory="Project manager.", allow_delegation=True)

task = Task(
    description="Write a 3-sentence summary of multi-agent coordination. "
                "Delegate research, then writing.",
    expected_output="A 3-sentence summary.",
    agent=manager,
)

# In hierarchical mode the manager goes ONLY in `manager_agent`,
# never in `agents`. Workers go in `agents`.
Crew(agents=[researcher, writer], tasks=[task],
     process=Process.hierarchical, manager_agent=manager).kickoff()
# AgentSonarListener auto-shuts down on CrewKickoffCompletedEvent.
```

After either script finishes, open the HTML report:

```bash
open "agentsonar_logs/$(cat agentsonar_logs/latest)/report.html"
```

## Status

Closed beta. Schema, public API, and output formats are stable for
design partners.

## License

Apache-2.0
