Metadata-Version: 2.4
Name: a2ea-py
Version: 0.2.0
Summary: Official Python client for the A2EA telemetry and attribution platform.
Author: AI Blockchain Ventures
Requires-Python: >=3.9
Description-Content-Type: text/markdown
Requires-Dist: requests>=2.31.0
Provides-Extra: langchain
Requires-Dist: langchain-core>=0.3; extra == "langchain"
Provides-Extra: crewai
Requires-Dist: crewai>=1.0; extra == "crewai"
Provides-Extra: openai-agents
Requires-Dist: openai-agents>=0.2; extra == "openai-agents"

# a2ea-py

Official Python SDK for sending A2EA telemetry to the backend.

## Install

```bash
pip install a2ea-py
```

## Quickstart

```python
from a2ea import A2EAClient

client = A2EAClient(
    api_key="your X-A2EA-API-Key"
)


@client.track_agent(task_type="data_processing")
def run_agent(payload):
    return process(payload)


client.flush(timeout=5)
client.close()
```

Send many completed tasks in one request when a worker already has a result batch:

```python
client.send_batch("data_processing", [
    {"event_id": "evt_101", "duration_ms": 840, "outcome": "success"},
    {"event_id": "evt_102", "duration_ms": 910, "outcome": "success"},
])
client.flush(timeout=5)
```

### Optional configuration

- `base_url`
  - Defaults to `https://a2ea-backend-526z2hkxna-uc.a.run.app`
  - Set this when testing locally against `http://127.0.0.1:4080`
- `agent_id`
  - Optional override if you want to assign a fixed agent identifier instead of the default host/process-based value

Example for local development:

```python
from a2ea import A2EAClient

client = A2EAClient(
    api_key="your X-A2EA-API-Key",
    base_url="http://127.0.0.1:4080",
    agent_id="agent_pipeline_014",
)
```

Use `A2EA_BASE_URL=https://a2ea-backend-526z2hkxna-uc.a.run.app` explicitly if your deployment pins runtime configuration through environment variables.

## Route work by A2EA shares

When workload management is on for a task type (Dashboard > Workload), A2EA publishes a routing share for each agent. A managing agent can pick the next agent with:

```python
agent_id = client.choose_agent("invoice_review", eligible_agent_ids=agents_allowed_for_this_task)
if agent_id is None:
    agent_id = my_default_router(task)  # workload management is off or no share applies
```

`choose_agent` only picks among `eligible_agent_ids`, so your own permission rules still decide which agents may run a task. Use `client.get_allocation(task_type)` to read the full policy, shares and history. In automatic mode, reading the shares applies the next change once the cooldown has passed.

## Credit shared jobs

When several agents work on one job, give each of their events the same `workflow_id`. A2EA splits the job's value among them (by share of events, time, or criticality) and the credits always add up to the job total.

```python
client.send_telemetry("invoice_intake", 8000, "success", workflow_id="invoice_6040")
review_client.send_telemetry("invoice_review", 30000, "success", workflow_id="invoice_6040", criticality=5)

split = client.get_workflow("invoice_6040")            # credits under the saved method
preview = client.get_workflow("invoice_6040", method="time_share")
totals = client.get_workflow_credits(days=30)           # each agent's credit across workflows
```

`criticality` is a whole number from 1 to 10 (1 when not set). Both fields are signed with the event. Batch events accept `workflow_id` and `criticality` keys too.

## Check a score

Every request that stores new events saves a score record with all its inputs. The score response includes `scoreRecordId` and `recordHash`.

```python
records = client.list_score_records(agent_id="agent_review")
record = client.get_score_record(records["records"][0]["id"])
for step in record["explanation"]:
    print(step["title"], step["value"])

result = client.verify_score_record(record["id"])  # recomputes the score, checks hash, chain and events
assert result["valid"]

# Re-check the hash yourself:
assert A2EAClient.score_record_hash(record["recordJson"], record["previousHash"]) == record["recordHash"]
```

## Weekly report

```python
report = client.get_weekly_report()                        # last complete week, as a dict
csv_text = client.get_weekly_report(week_start="2026-10-26", as_csv=True)
print(report["summary"]["netValueUsd"], report["reportHash"])
```

The JSON and CSV carry the same `reportHash`, report version and calculation version.

## Framework connectors

Thin connectors record agent framework runs with the same A2EA fields as the client: task type, agent, duration, success or failure, and a workflow ID so agents working on one job can share credit. Telemetry problems are logged and never raised into your agent. Each run gets a stable event ID, so it is never counted twice.

```bash
pip install "a2ea-py[langchain]"       # or [crewai], [openai-agents]
```

**LangChain**: a callback handler. One result per top-level run (chain, agent, model or tool); `track="tools"` records each tool call instead.

```python
from a2ea.integrations.langchain import A2EACallbackHandler

handler = A2EACallbackHandler(task_type="invoice_review", agent_id="invoice_chain")
chain.invoke(invoice, config={"callbacks": [handler], "metadata": {"a2ea_workflow_id": f"invoice_{invoice_id}"}})
```

**CrewAI** (1.x): an event listener. One result per task, with the agent's role as the agent ID; tasks in one kickoff share a workflow ID. `track="crews"` records whole kickoffs.

```python
from a2ea.integrations.crewai import instrument

instrument(task_type="market_research", task_types={"summary": "report_writing"})  # once
crew.kickoff()
```

**OpenAI Agents SDK**: a tracing processor. One result per agent that runs (including handoffs); a run's `group_id` becomes the workflow ID. `track="tools"` records function-tool calls.

```python
from a2ea.integrations.openai_agents import install

install(task_type="support_triage")  # once
await Runner.run(agent, "...", run_config=RunConfig(group_id=f"ticket_{ticket_id}"))
```

Per run, metadata keys `a2ea_task_type`, `a2ea_workflow_id` and `a2ea_criticality` override the defaults (LangChain also reads `a2ea_agent_id`). Runnable examples, which work offline with stand-in models, are in `examples/`.

## What the client sends

Each tracked function call produces:

- `eventId`
- `agentId`
- `timestamp`
- `durationMs`
- `taskType`
- `outcome`
- `verification`
- `workflowId`, `criticality` and `parentEventId` (the step this one came from), when set (and a per-result `agent_id` when one client reports several agents)

The verification payload is signed locally with HMAC-SHA256 using the workspace API key so the backend can verify the telemetry envelope.

## Reliability behavior

- submission runs in a background thread
- transient failures are retried with backoff
- exhausted events are retained in a local fallback buffer

## Local package test commands

```bash
python -m unittest discover tests
python -m build
pip install dist/a2ea_py-0.1.0-py3-none-any.whl
```
