Getting Started

Download & Install

Install agentflight-mcp from PyPI (imports stay import agentflight and the CLI stays agentflight), or run it straight through uvx without adding packages to your project.

pip install agentflight-mcp && agentflight doctor

First Flight Run

Verify every subsystem locally or run the end-to-end demo without API keys:

agentflight demo
agentflight lint
agentflight benchmark

Choosing a Recording Mode

• record — Executes live LLM/tool calls and writes an atomic SHA-256 cassette with API keys scrubbed.
• replay — Serves recorded responses offline in microseconds (`$0.00` spend) and fails immediately on structural drift.
• auto (default in pytest) — Records on the first missing cassette, then replays deterministically in CI.

Self-Diagnostics (`agentflight doctor`)

Run built-in diagnostics covering Python 3.10+, zero-dep stdlib core imports, lazy pytest11 entry point registration, SHA-256 redaction integrity, and circuit-breaker loop trapping:

agentflight doctor --json

OpenAI & Compatible Clients

Wrap synchronous or async OpenAI, AsyncOpenAI, OpenRouter, vLLM, or OpenAny clients in one line—recording both standard responses and stream=True delta chunks.

from openai import OpenAI
from agentflight import wrap_openai, TrajectoryCircuitBreaker

client = wrap_openai(OpenAI(), guard=TrajectoryCircuitBreaker(max_cost_usd=0.25))
resp = client.chat.completions.create(model="gpt-4o-mini", messages=[{"role": "user", "content": "Health?"}])
client.agentflight_recorder.save("tests/cassettes/health.flight.json")

Anthropic Messages & Streaming

Use wrap_anthropic(Anthropic(), ...) to capture synchronous, async, and streamed Claude messages with coalesced tool-use blocks and token telemetry.

Callbacks & Decorators (`wrap_callable`)

Instrument any Python tool function with @wrap_callable or pass FlightCallbackHandler directly into LangChain, LangGraph, or custom agent loops.

Canonical JSON SHA-256 Hash Chain

Every TrajectorySpan hashes canonical sorted JSON (sort_keys=True) chained from GENESIS_HASH via prev_hash -> step_hash. Reordered Python dictionary keys never cause false negatives, whereas silent edits trigger CassetteTamperError.

Automatic PII & Secret Scrubbing

FlightRecorder(..., redact_secrets=True) scrubs OpenAI, Anthropic, GitHub, AWS keys, Bearer tokens, JWTs, Basic Auth URLs, and auth headers before computing SHA-256 digests so cassettes can be safely committed to Git.

Deterministic Stubs (`CassetteReplayer`)

from agentflight import CassetteReplayer

replayer = CassetteReplayer("cassettes/incident_8841.flight.json", strict_order=True)
query_metrics = replayer.stub_tool("prometheus.query_metrics", kind="mcp_rpc")
result = query_metrics(window="15m", service="payment-api")

Auto-Registered Pytest Plugin (`pytest11`)

Installing agentflight-mcp automatically registers built-in fixtures (flight_recorder, flight_guard, flight_session) plus @pytest.mark.flight_cassette(...).

import pytest

@pytest.mark.flight_cassette("tests/cassettes/checkout.flight.json")
def test_checkout_deterministic(flight_session, flight_guard):
    search = flight_session.stub("search_catalog", fallback_fn=lambda q: {"hits": [q]})
    assert search(q="router")["hits"] == ["router"]

Behavioral Trajectory Diffing

Compare baseline and candidate cassettes across model or prompt upgrades using Levenshtein/LCS sequence distance plus token/cost/error deltas:

agentflight diff baseline.flight.json candidate.flight.json --similarity-threshold 0.85

GitHub Actions Pre-Flight Gate

agentflight ci --schema-file tools.json --cassette tests/cassettes/run.flight.json --fail-on-smell --github-summary

Loop & Ping-Pong Detection

TrajectoryCircuitBreaker inspects actor + canonical argument digests on every turn to stop exact identical repeats, two-step A -> B -> A -> B ping-pong loops, three-step cycles, and consecutive error storms.

Thread-Safe Token & USD Spend Caps

Enforce hard limits globally across worker threads via with breaker.reserve_budget(estimated_usd): ... so parallel tool invocations cannot overshoot budget ceilings.

from agentflight import TrajectoryCircuitBreaker

guard = TrajectoryCircuitBreaker(max_identical_repeats=3, max_ping_pong_cycles=2, max_total_tokens=25000, max_cost_usd=0.50)

Tools, Resources & Prompts

Hosted Vercel HTTP vs Local Stdio

claude mcp add --transport http agentflight https://agentflight-tau.vercel.app/mcp

AAIB / NTSB Incident Post-Mortems

Generate structured Markdown or JSON failure reports pinpointing root-cause loops, cost explosions, and exact failing tool steps from any saved .flight.json file:

agentflight post-mortem crash_incident.flight.json --output POST_MORTEM.md

W3C TraceContext / OpenTelemetry Export

Export cassettes into gen_ai.* semantic-convention spans ready for StackPulse, Jaeger, Datadog, or Grafana Tempo:

agentflight export-otel incident_8841.flight.json --output spans.json

Budget Shield & Streaming

OpenAnyProvider adds strict domain isolation, pre-execution cost ceiling checks, SSE streaming, and remote MCP schema discovery across 128+ models with transparent attribution when querying *.openany.si.

OpenAny CLI Commands

agentflight openany models
agentflight openany lint-tools
agentflight openany replay --cassette paxos.flight.json

TrajectoryDivergenceError

Raised during replay mode when candidate code calls a different actor or passes modified parameter keys/values compared to the recorded baseline cassette. Inspect exc.missing_keys, exc.extra_keys, and exc.changed_values, or re-record intentionally with pytest --flight-mode=record.

CassetteTamperError

Raised by verify_cassette() when any step payload, index, or hash link in a .flight.json file is altered or truncated. Re-record cleanly from a trusted run or validate with agentflight doctor.

License & Provenance

Created and maintained by Akandwanaho Ivan Alvin under the permissive MIT License. Source hosted on GitHub (`codesphere-u/agentflight`).