Python SDK

trefur-observe — observability and safety telemetry for AI agents, with auto-instrumentation for Anthropic, OpenAI, LangChain, CrewAI, and friends.

Install

pip install trefur-observe

Optional extras pre-install the upstream SDKs:

pip install "trefur-observe[anthropic]"
pip install "trefur-observe[openai]"
pip install "trefur-observe[langchain]"
pip install "trefur-observe[crewai]"
pip install "trefur-observe[all]"

Init

import os
from anthropic import Anthropic
from trefur_observe import TrefurObserve, patch_anthropic

TrefurObserve.init(
    api_key=os.environ["TREFUR_API_KEY"],
    agent={"name": os.environ.get("TREFUR_AGENT_NAME", "my-agent")},
)

client = Anthropic()
patch_anthropic(TrefurObserve.get_instance())

Flushing on exit

Telemetry is buffered and shipped by a background flush every 5s. TrefurObserve.init() registers an atexit hook that flushes on a normal interpreter exit (including after an unhandled exception), so a short script's first trace still lands. For scripts that may raise or call a hard os._exit(), wrap your run in try/finally and call shutdown() to flush explicitly:

observe = TrefurObserve.init(api_key=os.environ["TREFUR_API_KEY"])

try:
    # ... your agent code ...
    client.messages.create(...)
finally:
    observe.shutdown()  # flush buffered telemetry (also runs on exception)

Configuration options

Option	Type	Default	Notes
`api_key`	`str`	required	`trf_obs_*`
`endpoint`	`str`	`https://observe.trefur.com`	Ingest endpoint override
`debug`	`bool`	`False`	Verbose logging
`batch_size`	`int`	`10`	Buffer size before auto-flush
`flush_interval`	`float`	`5.0`	Periodic flush interval, in seconds
`auto_instrument`	`bool`	`True`	Auto-patch known SDKs
`sql`	`bool`	`False`	Opt-in auto-patch of psycopg2 / asyncpg / SQLAlchemy
`capture_subprocess`	`bool`	`True`	Patch `subprocess` for CLI captures
`subprocess_filter`	`list[str]`	all	Limit to specific binary names
`agent`	`dict`	detected	Explicit agent identity

Framework adapters

from trefur_observe import (
    patch_anthropic,
    patch_openai,
    patch_gemini,
    patch_langchain,
    patch_crewai,
    patch_claude_agent_sdk,
    patch_playwright_sdk,
    patch_selenium,
)

Generic in-process instrumentors

from trefur_observe import TrefurObserve, observe

TrefurObserve.init(api_key=os.environ["TREFUR_API_KEY"])

# Patch HTTP libs, subprocess, and pathlib I/O in one call.
observe(http=True, fs=True, exec=True)

Axis	Modules patched	step_type
`http`	`requests`, `aiohttp`, `urllib.request`	`http_call`
`exec`	`subprocess.{run, Popen, check_output, check_call}`, `os.system`, `asyncio.create_subprocess_*`	`exec_subprocess` / `cli_exec`
`fs`	`pathlib.Path.{read,write}_{text,bytes}` and `unlink`	`file_read` / `file_write` / `file_delete`

Environment variables

Canonical list: env-vars reference.

Troubleshooting

Symptom	Fix
`patch_anthropic` appears to do nothing.	Make sure `TrefurObserve.init(...)` ran first. Set `debug=True` to surface failures.
async code: spans don't nest.	Use `parent_run_scope(...)` from `trefur_observe.context` to propagate the parent run id across `asyncio` task boundaries.
I want to mute the SDK in tests.	Set `TREFUR_DISABLED=true` in the environment. Short-circuits every emit.

Source + tests live in trefur-ai/trefur-sdks under python/observe-python. License: Apache 2.0.