bekchiai 0.1.0

BekchiAI SDK — observe, analyze, and contain your AI agents.

bekchiai

Observe, analyze, and contain your AI agents.

bekchiai is the Python SDK an agent project installs to connect to the BekchiAI platform. It auto-captures tool calls, LLM calls (with tokens & cost), URL requests, plans, and outcomes; ships them efficiently over a non-blocking batched transport; and can receive remote stop/pause commands. The core is dependency-free and framework-agnostic — every framework is an opt-in extra.

Install

uv add bekchiai          # recommended (uv project)
uv pip install bekchiai  # into the active environment
pip install bekchiai     # plain pip also works

uv add "bekchiai[crewai]"   # with a framework extra
uvx bekchiai doctor         # run the CLI once, no install

Quickstart

import os
from bekchiai import BekchiAI

lens = BekchiAI(
    agent_id="agt_8f3k2m91",
    api_key=os.environ["BEKCHIAI_API_KEY"],
    env="production", host="prod-worker-03", region="us-east-1",
)

agent = lens.wrap(my_agent)        # auto-captures tools, LLM calls, URLs, plans
agent.run("export the weekly dataset")

Three integration depths

1. lens.wrap(agent) — richest; hooks tools + LLM + plan/outcome, and dispatches to a framework adapter (Claude/Anthropic, LangChain, LangGraph, CrewAI, OpenAI) when it recognizes one.

2. Decorators / context managers — manual but precise:

@lens.tool                                  # → tool_call / tool_result
def search(query: str) -> list[str]: ...

with lens.session(task="export the weekly dataset"):
    with lens.plan() as p:
        p.state("fetch then upload", tools=["search", "upload"])
    lens.llm_call(model="claude-opus-4-8", input_tokens=1200, output_tokens=300)
    search("rows")

3. Raw lens.emit(event) / OTel exporter — bring-your-own telemetry:

lens.emit({"type": "url_access",
           "payload": {"method": "GET", "url": "https://api.example.com"}})

# Or export events as OpenTelemetry spans instead of POSTing to the API:
from bekchiai.integrations.otel import OTelSink
lens = BekchiAI(agent_id="...", sink=OTelSink())

Token capture & cost

Wrapping a provider client records model, input_tokens, output_tokens, and computes cost_usd from a built-in, updatable price table:

from anthropic import Anthropic
client = lens.wrap(Anthropic())     # every messages.create → llm_call (+cost)

HTTP capture (URL Monitor / exfiltration detection)

lens.instrument_http()   # patches `requests` → emits url_access per request

If a local allowlist policy is in enforce mode, off-allowlist hosts are blocked before the request and a denied event is emitted.

Remote control (Stop / Pause / Revoke)

lens = BekchiAI(agent_id="...", api_key="...", enable_control=True)
# In the run loop / inside tools, honor remote commands:
lens.checkpoint()        # raises AgentStopped on stop; blocks while paused

Two stop scopes are honored: session (halt the current run) and quarantine/revoke (also refuse new sessions until resumed).

Local policy enforcement (defense in depth)

from bekchiai import BekchiAI, Policy

lens = BekchiAI(
    agent_id="...", api_key="...",
    policy=Policy(allowed_hosts=["*.example.com"], allowed_tools=["search"], enforce=True),
)

Redaction

Secret-shaped values (API keys, bearer tokens, JWTs, high-entropy blobs) and sensitive keys (password, authorization, …) are redacted before events leave your environment. Use redact_salt= to get stable salted tokens for grouping without storing raw PII.

CLI

bekchiai doctor        # verify key, connectivity, clock skew; send a test event
bekchiai test-event    # send a single test event

The event contract

Every event matches the shared schema in 00-shared-event-schema.md. No integration may invent new event types — adapters only call the core builders. A contract test (tests/test_contract.py) validates emitted events against the schema.

type	when
session_start	a task begins
plan	the agent states intent
llm_call	a model call completes (model, tokens, cost)
tool_call / tool_result	a tool is invoked / returns
url_access	an HTTP request
denied	an action blocked by policy
boundary_attempt	the agent probes a containment boundary (outbound)
attack_attempt	an attack targeting the agent (inbound)
outcome	a task ends

Development

uv venv && uv pip install -e ".[dev]"
uv run pytest
uv run ruff check .
uv run mypy src/bekchiai
uv build

(Plain python -m venv .venv && pip install -e ".[dev]" works too.)

License

MIT