precipiq 0.1.0

Precipiq Python SDK — track AI agent decisions and measure their financial consequences

precipiq — Python SDK

Track AI agent decisions and measure their financial consequences.

precipiq records the decisions your AI agents make, chains each one into an append-only, tamper-evident ledger, and later links those decisions to the financial events they caused — so you can attribute revenue, cost, and liability back to the agent that drove them.

Installation

pip install precipiq

Optional framework adapters are opt-in extras (install only what your stack needs):

pip install "precipiq[langchain]"   # LangChain callback handler
pip install "precipiq[crewai]"      # CrewAI callback handler (Python < 3.14)

Requires Python 3.11+. The core SDK has no upper Python bound; the crewai extra carries a < 3.14 ceiling (a CrewAI transitive constraint), so on Python 3.14+ install precipiq[langchain] or pin the adapter yourself.

Log your first decision

# demo.py
from precipiq import Precipiq

# ``enable_batching=False`` for the demo so the call ships synchronously and
# ``log_decision`` returns the server receipt instead of ``None``. In
# production, leave batching on for throughput and call ``pq.flush()`` at
# shutdown.
pq = Precipiq(
    api_key="pq_test_demo_key_REPLACE_ME",
    enable_batching=False,
)

receipt = pq.log_decision(
    agent_id="pricing-bot",
    action_type="discount_offer",
    inputs={"customer_id": "cust_123", "tier": "gold"},
    outputs={"discount_pct": 15},
    confidence=0.82,
    human_in_loop=False,
)
# ``receipt`` is a dict keyed by the server's snake_case schema.
print("decision id:", receipt["decision_id"])
print("chain hash:", receipt["hash"])

Wrap a real function with track

In production you rarely call log_decision directly. Instead, wrap the function that makes the decision — the SDK captures inputs, outputs, and any exception automatically.

@pq.track(agent_id="pricing-bot", action_type="discount_offer")
def compute_discount(customer_id: str, tier: str) -> dict:
    pct = 15 if tier == "gold" else 5
    return {"discount_pct": pct}

compute_discount("cust_456", "silver")

Link a decision to a financial outcome

When the customer is charged (Stripe webhook, internal event, etc.), link the decision to the resulting financial event so the AI P&L can attribute dollars.

pq.link_outcome(
    receipt["decision_id"],
    "fev_abc",                      # the financial_event_id from your integration
    correlation_strength=1.0,       # 0..1 — strongest when directly caused
    link_type="revenue",            # "revenue" | "cost" | "liability" | "neutral"
)

Check your AI P&L

pnl = pq.get_ai_pnl(start="2026-04-01", end="2026-04-30")
if pnl is not None:
    print(pnl["total_revenue_attributed"], pnl["currency"])
    print(pnl["net_ai_impact"])

Async client

Every method has an async counterpart on AsyncPrecipiq, which doubles as an async context manager so the buffer is flushed on exit.

from precipiq import AsyncPrecipiq

async with AsyncPrecipiq(api_key="pq_live_...") as pq:
    await pq.log_decision(
        agent_id="refund-bot",
        action_type="approve",
        inputs={"ticket_id": "t1"},
        outputs={"refund_amount": 100},
        confidence=0.95,
    )

Framework adapters

The adapters are import-time optional — importing the SDK core never imports LangChain or CrewAI. Import the adapter module yourself once the matching extra is installed:

from precipiq import Precipiq
from precipiq.integrations.langchain import PrecipiqLangChainCallback
from precipiq.integrations.crewai import PrecipiqCrewAICallback

pq = Precipiq(api_key="pq_live_...")
# Pass the callback into your LangChain / CrewAI run to record each step.
# ``agent_id`` is required — it is stamped on every decision the callback emits.
handler = PrecipiqLangChainCallback(pq, agent_id="pricing-bot")

Graceful degradation

If the Precipiq API is unreachable, the SDK does not raise through your application. Decisions it could not deliver are appended to a rotating local fallback log (~/.precipiq/fallback.log by default) so an operator can replay them later.

Public surface

• Precipiq / AsyncPrecipiq — sync and async clients.
• Methods: log_decision, link_outcome, get_ai_pnl, track, flush.
• Errors: PrecipiqError, PrecipiqAPIError, PrecipiqAuthError, PrecipiqTransportError.
• Adapters: precipiq.integrations.langchain, precipiq.integrations.crewai.

Links

• Quickstart: https://docs.precipiq.com/quickstart
• Documentation: https://docs.precipiq.com
• Source: https://github.com/lexwhiting/precipiq

Local development

cd packages/sdk-python
poetry install
poetry run pytest

License

MIT