usehusk

Attribute AI usage to customers from Python requests, agents, and background jobs. Python 3.11+. The SDK core has no runtime dependencies.

Install

pip install usehusk==0.1.0a1

This is an alpha, so pip needs the version (or --pre) to install it. Install your provider or agent framework separately.

Let a coding agent integrate it

The package ships step-by-step instructions for coding agents (Claude Code, Cursor, Codex and the like) in usehusk/AGENTS.md. After installing, give your agent a prompt such as:

Integrate the usehusk package into this project to attribute our AI usage to our customers. Follow the instructions in its AGENTS.md; find the file with python -c "import usehusk, pathlib; print(pathlib.Path(usehusk.__file__).with_name('AGENTS.md'))".

Wrap a provider client

from openai import AsyncOpenAI
from usehusk import Husk

husk = Husk(api_key="husk_...", customer={"id": "customer_123"})
client = husk.wrap(AsyncOpenAI())
response = await client.chat.completions.create(
    model="gpt-4.1-mini",
    messages=[{"role": "user", "content": "Hello"}],
)

Wrap once per customer/request, then use the returned client normally. The same API accepts sync/async OpenAI and Anthropic clients, google.genai.Client, and PydanticAI models. Original clients remain unchanged. No manual capture is needed.

Streaming produces one event on exhaustion, close, or context exit. Partial usage is marked incomplete; streams are never drained just for telemetry. OpenAI chat streams request stream_options.include_usage=True unless you explicitly set it to False. Provider responses and chunks are returned unchanged; stream/client objects are proxies. Close them using the provider's usual API.

Only the listed generation methods are instrumented. Raw HTTP response helpers, Gemini chat sessions/automatic tool loops, embeddings, images, audio, and realtime APIs are not covered. Use explicit capture for supported responses on other paths. Wrap either the provider or its PydanticAI model, not both.

Wrap a PydanticAI model

from pydantic_ai import Agent
from usehusk import Husk

husk = Husk(api_key="husk_...", customer={"id": "customer_123"})
model = husk.wrap(existing_model)
agent = Agent(model)

existing_model is your PydanticAI model object. Run the agent normally; Husk captures each model interaction, including calls during tool loops and streaming. The original model remains unchanged. Other requests can wrap it with their own customer identity if the provider supports concurrent use.

Create Husk after authentication or from a job's stored identity. There is no global initialization, lifecycle hook, or required context manager. Compatible clients share background delivery within the worker process.

Add optional attribution

husk = Husk(
    api_key="husk_...",
    customer={"id": "company_123", "name": "Acme"},
    user={"id": "person_456", "name": "Jane"},
    product={"id": "support"},
    session_id="conversation_789",
    trace_id="turn_001",
    environment="prod",
    labels=["support", "experiment"],
)
model = husk.wrap(existing_model, feature="answer_question")

Only the API key and customer ID are needed. HUSK_API_KEY can supply the key. Customer and user stay fixed for the instance; product, feature, event, correlation IDs, environment, labels, and scalar properties can be supplied per operation. HUSK_ENVIRONMENT supplies the environment when omitted at construction. Labels are a list of strings; an operation override replaces the list. Passing None clears either field.

Capture a direct provider response

For a call made through an unwrapped client:

response = await provider_call()  # Your existing provider call.
husk.capture(response=response, feature="extraction")

Supported responses include OpenAI Chat/Responses, Anthropic Messages, Gemini, OpenRouter, PydanticAI ModelResponse, and LangChain AIMessage. Pass provider="openrouter" for OpenRouter responses that resemble OpenAI. Husk.wrap() captures supported provider calls automatically; explicit capture is optional for calls outside those wrappers.

For custom accounting:

husk.capture(provider="custom", model="internal-model", input_tokens=120, output_tokens=30)

Capture each model call once. Do not also submit aggregate agent usage for calls already captured by the adapter. To report the same saved result again, explicit capture accepts an optional idempotency_key; backend enforcement is required.

Delivery and diagnostics

The default endpoint is https://api.usehusk.com/api/v1/ai-spend/sdk-usage. base_url or HUSK_BASE_URL overrides the origin for development ingestion. A background thread sends bounded batches with urllib. Prompts, response content, and tool arguments are not collected. Names, environment, labels, and properties are sent as supplied.

On macOS, use the default spawn start method for multiprocessing. If your app explicitly uses os.fork(), Python documents setting no_proxy=* before forking to avoid an unsafe system-proxy lookup in urllib. This disables proxies and does not make other libraries safe to fork. See the Python urllib warning.

stats = husk.stats()
print(stats["last_capture_error"])
print(stats["delivery"])

# Optional checkpoint for scripts or a runtime about to freeze:
acknowledged = husk.flush(timeout=10)

Capture returns a detached queued event or None; queueing does not confirm server acceptance. Delivery is best effort. Queue overflow, exhausted retries, or abrupt process termination can lose events. Ordinary requests need no flush. For an async checkpoint, use await asyncio.to_thread(husk.flush, timeout=10).

This is a development alpha. Live ingestion acceptance, pricing, and backend deduplication remain unverified.

Read the customer guide for FastAPI dependencies, middleware, approval/resume, streaming, jobs, and troubleshooting. The ingestion contract describes backend requirements; the verification record describes test coverage.

Development

uv sync --locked
uv run python -m unittest discover -s tests -v
uv run ruff check .
uv run ruff format --check .
uv run pyright
uv run sh scripts/generate_types.sh
uv build

Offline tests use synthetic provider responses and local/mock ingestion. The opt-in scripts/live_smoke.py makes one paid provider call and checks ingestion acknowledgement; it does not verify pricing. Set HUSK_LIVE_SMOKE=1, a test key and development origin, HUSK_LIVE_PROVIDER, HUSK_LIVE_MODEL, and provider credentials.