aforo-agent-metering

Meter AI agent runs from Python: record each capability invocation and reasoning step as an AI_AGENT usage event, buffered and batched to Aforo. For agent runtimes such as LangChain, LlamaIndex, CrewAI, AutoGen and FastAPI-hosted agents. Python sibling of @aforoai/agent-metering (Node).

Version: 0.3.2 · Changelog · User guide

Install

pip install aforo-agent-metering
# pick an async HTTP client (optional — stdlib urllib is the fallback):
pip install "aforo-agent-metering[aiohttp]"
pip install "aforo-agent-metering[httpx]"

From source:

git clone https://github.com/aforoai/SDKs.git
cd SDKs/aforo-metering-sdks/python-agent
pip install -e .

Quickstart

import asyncio
import os
from aforo_agent_metering import AforoAgentClient

async def main():
    client = AforoAgentClient(
        tenant_id="tenant_xxx",
        product_id="prod_ai_001",
        api_key=os.environ["AFORO_API_KEY"],
        # ingestor_url defaults to https://api.aforo.ai
    )
    await client.start()  # begin the periodic flush task

    client.record_capability(   # synchronous: buffers the event
        capability_name="summarize_email",
        agent_id="agt_001",
        customer_id="cust_42",
        session_id="sess_1",
        input_tokens=320,
        output_tokens=84,
        execution_duration_ms=510,
    )

    await client.shutdown()     # final flush

asyncio.run(main())

Events POST to https://api.aforo.ai/v1/ingest/batch as {"events": [...]} with X-API-Key: <api_key> and X-Tenant-Id: <tenant_id>. No Authorization header is sent.

Decorator

from aforo_agent_metering import AforoAgentClient, wrap_capability_handler

client = AforoAgentClient(...)

@wrap_capability_handler(client, capability_name="summarize_email")
async def summarize_email(text: str, *, agent_id: str, session_id: str, customer_id: str):
    ...

The decorator times the handler, records one invocation when it finishes, and re-raises whatever the handler raised. Status: returned → SUCCESS; TimeoutError → TIMEOUT; any other exception → ERROR; asyncio.CancelledError / KeyboardInterrupt / SystemExit → CANCELLED. The handler must be async def.

Configuration

Execution status

execution_status is trimmed and upper-cased. Accepted values: SUCCESS, PARTIAL, TIMEOUT, ERROR, VALIDATION_FAILED, FAILED, FAILURE, CANCELLED, PENDING, BLOCKED, HITL_REQUIRED (exported as EXECUTION_STATUSES). A blank or unknown value is logged and left off; the event is still sent. OUTCOME_BASED rate plans bill each event at the weight set for its status — see the repository README.

Retries and dropped events

Retry: 5xx, 408, 429 and network errors, with 1s / 2s backoff; a 429 waits for its Retry-After. Any other 4xx is not retried.

record_capability and record_step never raise for event content. Every event the SDK cannot deliver is counted in client.dropped_count, WARN-logged, and passed to on_drop(events, reason). Dropped events keep their idempotency keys.

Wire format

Each event carries top-level customerId, metricName (ai_agent.capability_invocations or ai_agent.steps), quantity 1, occurredAt, idempotencyKey (agent:<uuid4>, minted once when the event is created), productType, agentId, sessionId, executionDurationMs, optional executionStatus, and metadata with capability_name (snake_case — the key the ingestor maps to the per-capability billing dimension), token counts and sdkVersion.

The endpoint path and body shape are checked against contract/ingest-contract.json by this package's tests.

Walk me through it

Install → construct → record → confirm the event in Aforo: USER_GUIDE.md. A runnable example is in examples/hello_agent.

What this doesn't cover

This SDK emits usage events. It does not send session heartbeats, check entitlements or quotas, or price anything — rate plans and per-capability pricing are configured in the Aforo console. record_capability / record_step only buffer; without await start() or an explicit await flush() nothing is sent until the buffer reaches flush_count inside a running event loop.