agenthog 0.1.2

Official Python SDK for AgentOS — observability, runtime, and security for AI agents.

agenthog (Python SDK for AgentOS)

Official Python SDK for AgentOS — observability, runtime, and security for AI agents. Currently at v0.1.0.

The package is named agenthog because AgentHog is the developer-facing observability surface of the AgentOS platform; the SDK ships those primitives under that name. (The platform itself is still called "AgentOS" and the GitHub repo is still agentos-python.)

Install

pip install agenthog
# auto-instrumentation extras (each is opt-in):
pip install "agenthog[openai]"
pip install "agenthog[anthropic]"
pip install "agenthog[langchain]"
pip install "agenthog[otel]"
# or everything at once:
pip install "agenthog[all]"

Requires Python 3.10 or newer.

First trace (5 lines)

import agenthog

agenthog.init(api_key="agops_…", agent_id="my-agent")

with agenthog.start_task_run() as run:
    # Your agent logic here. Auto-instrumented LLM / tool calls land
    # automatically; manual events go through `client.log_*(...)`.
    ...

That's it. Every supported integration emits the same agent.* event shape so your traces look the same regardless of which framework you end up calling.

Quickstart with auto-instrumentation

import os
import agenthog
from agenthog.integrations import openai as openai_integration
from openai import OpenAI

agenthog.init(api_key=os.environ["AGENTOS_API_KEY"], agent_id="weather-agent")
openai_integration.install()  # one line — every chat.completions.create is now traced

oai = OpenAI()
with agenthog.start_task_run() as run:
    resp = oai.chat.completions.create(
        model="gpt-4o-mini",
        messages=[{"role": "user", "content": "Weather in Paris?"}],
    )
    print(resp.choices[0].message.content)

Each integration ships an install() / uninstall() pair. They are transparent: original return values pass through and the original exceptions re-raise unchanged.

Configuration

The SDK reads environment variables first; explicit init(...) args override. Full set per docs/spec/05-sdk-contract.md:

Field	Env var	Default
api_key	AGENTOS_API_KEY (required)	—
endpoint	AGENTOS_ENDPOINT	https://api.theagentos.space
environment	AGENTOS_ENV	unset
agent_id	AGENTOS_AGENT_ID	unset (must be set here or at start_task_run)
project_id	AGENTOS_PROJECT_ID	resolved server-side from API key
capture_content	—	True (set False to strip input / output / tool.input / tool.output)
flush_interval_ms	—	250
flush_batch_size	—	100
buffer_max_size	—	10000 (drop oldest on overflow)
disable	AGENTOS_DISABLE	False (1 / true ⇒ fast no-op)
kind	—	"sync" (or "async" for async-native hosts)

disable=True makes every public method a fast return — useful in unit tests that don't want a mock endpoint.

Integrations

Provider	Module	Patches
OpenAI	agenthog.integrations.openai	chat.completions.create (sync + async, streaming + non-streaming, tool calls)
Anthropic	agenthog.integrations.anthropic	messages.create (with stream=True) and messages.stream
LangChain	agenthog.integrations.langchain	BaseCallbackHandler registered globally; one event per LLM call regardless of token count
OpenTelemetry passthrough	agenthog.integrations.opentelemetry	TEE: translates OTel spans to agent.* events; optional forward_to= keeps your collector

All integrations honor capture_content=False and emit exactly one event per logical operation — streaming aggregations close into a single agent.llm_call event with the full output and final usage.

Manual logging

Use client.log_* for anything not auto-instrumented:

client = agenthog.get_default_client()
assert client is not None

client.log_eval("faithfulness", score=0.92, evaluator="llm")
client.log_business_event("checkout_completed", revenue=29.0, currency="USD")
client.log_security_alert("prompt_injection", severity="high", action_taken="blocked")
client.log_handoff("researcher_agent", reason="needs deep search")
client.log_retrieval("vector_kb", query="...", top_k=5, results_count=3)

All inherit the active identity context. All validate locally; failed validation drops the event with a WARN — your code never raises.

Flags

variant = client.flag("use_gpt4_for_support", default=False, user_id="user_42")

The SDK polls GET /v1/flags every 60 s and evaluates locally (priority-ascending, first-match-wins; percentage rollouts use a stable hash so the same user stays in or out across calls). Each flag() call also emits an agent.flag_check event so you can correlate flag state with downstream behavior.

Routing

decision = client.route(goal="answer_well", context={"topic": "support"})
# decision -> {"arm_id": "...", "model": "...", "decision_span_id": "..."}

# Later, after the model answers, close the bandit's loop:
client.route_outcome(decision["decision_span_id"], reward=1.0, success=True, feedback_kind="auto")

Inside an AgentRun container (AGENTOS_RUNTIME=1), route and route_outcome go through the syscall mediator sidecar over a Unix socket — same SDK code path, no env var to flip.

Privacy controls

• capture_content=False strips input, output, tool.input, tool.output, retrieval.query, retrieval.documents from every emitted event. Token counts, model names, durations, costs, and identity fields still flow.
• Redactors run before send: client.add_redactor(lambda e: ...). Redactors must be deterministic and fast; if one raises, it's skipped (with a WARN) and the event still ships.
• The SDK detects email / E.164 phone patterns in user_id and emits a one-time WARN per id so you can pseudonymize before they hit the wire.

SDK logs

Internal SDK logs live under the agenthog.sdk namespace, default level WARNING. Silence or amplify via standard Python logging:

import logging
logging.getLogger("agenthog.sdk").setLevel(logging.DEBUG)

Sub-loggers per submodule: agenthog.sdk.transport, agenthog.sdk.context, agenthog.sdk.runtime, agenthog.sdk.integrations.<name>, agenthog.sdk.flags.

Failure-mode contract

Per docs/spec/05-sdk-contract.md "Failure modes": the SDK never raises out of public methods. A misconfigured endpoint, a network outage, a malformed event — every error flows to the internal logger and the corresponding metric counter, never to your application.

Counters live on client.metrics:

• events_sent — successfully ingested
• events_dropped — buffer overflow
• events_validation_failed — schema rejection (local)
• events_retried — retried at least once before success or permanent-fail
• events_failed_permanent — gave up after max_retries

Hook into them via client.on_metric(lambda name, delta: ...).

Versioning (SemVer)

• Patch (0.1.0 → 0.1.1): new event types, new optional fields, bug fixes.
• Minor (0.1.0 → 0.2.0) while still in 0.x: any breaking change to public methods.
• Major (0.x → 1.0.0): graduation to stable.
• Once on 1.x, breaking changes bump major.

The __version__ constant in agenthog/__init__.py and the version field in pyproject.toml are kept in sync; the publish workflow refuses to ship if the tag and __version__ disagree.

Local development

uv sync
uv run pytest -q
uv run mypy agenthog/
uv run ruff check . && uv run ruff format .

Integration tests

The end-to-end test in tests/integration/test_e2e.py runs against a live AgentOS platform. It is skipped automatically unless both env vars are set:

export AGENTOS_TEST_ENDPOINT=https://api.theagentos.space
export AGENTOS_TEST_API_KEY=agops_…
# Optional override:
export AGENTOS_TEST_AGENT_ID=e2e-test-agent

uv run pytest tests/integration/test_e2e.py -v

The test mocks the OpenAI client (no real API costs) but every event flows through the actual transport, hits the actual ingestion API, and is read back via GET /v1/traces/:trace_id.

Publishing

This repo publishes via Trusted Publishing (OpenID Connect) — no API tokens stored anywhere. Two workflows:

• publish.yml — on every v* tag, ships to production https://pypi.org/project/agenthog/.
• publish-test.yml — manual workflow_dispatch, ships to https://test.pypi.org/project/agenthog/. Use this to dry-run a release before tagging the real one.

One-time setup (per index — TestPyPI and PyPI are independent):

1. Register the project on the index (one manual upload is enough to claim the agenthog name on each).
2. On the project's "Publishing" settings page, add a Trusted Publisher with:
   • Owner: your GitHub user / org.
   • Repository: agentos-python.
   • Workflow: publish.yml (PyPI) or publish-test.yml (TestPyPI).
   • Environment: pypi or testpypi respectively.
3. In this repo's GitHub settings → Environments, create the matching environment(s). No required reviewers — the OIDC handshake is the gate.

Recommended release flow:

# 1. Dry-run on TestPyPI: Actions → publish-test → Run workflow → main
pip install -i https://test.pypi.org/simple/ \
  --extra-index-url https://pypi.org/simple/ \
  agenthog==0.1.0
# … smoke-test in a fresh venv …

# 2. Tag for production
git tag v0.1.0
git push origin v0.1.0

The publish workflow builds, verifies the tag matches __version__, and publishes to PyPI.

main branch protection requires the test workflow to pass before merge.

Spec

The four spec files this SDK honors live under docs/spec/:

• 05-sdk-contract.md — the SDK contract.
• 03-event-schema.md — emitted event types.
• 02-identity-model.md — project / agent / task_run / span hierarchy.
• 60-api-conventions.md — error envelope, headers, idempotency.

When spec and code disagree, fix the spec first. Same rule as the platform monorepo.

License

Apache License 2.0. The published agenthog package on PyPI ships under the same terms — including the explicit patent grant and the standard "AS IS" disclaimer in section 7 of the license. The package contains no code from the AgentOS platform monorepo; only the SDK client surface defined in docs/spec/05-sdk-contract.md.