kintic-sdk 0.1.10

Context logging for AI agents

Kintic Python SDK (kintic-sdk)

Ship every agent decision — with full context, policy, and outcomes — to your Kintic dashboard.

Install

From PyPI:

pip install kintic-sdk

That installs only the core SDK plus requests. Framework and LLM packages are optional so installs stay fast and you avoid dragging in CrewAI/LangGraph/etc. when you only use OpenAI.

Want everything in one shot (all framework hooks + common LLM clients):

pip install "kintic-sdk[all]"

Pick what you use (smaller, fewer version conflicts):

pip install "kintic-sdk[langgraph,crewai,groq]"
pip install openai anthropic   # provider SDKs for kintic.patch()

Upgrading an existing environment:

python3 -m pip install --upgrade kintic-sdk

On macOS (and some Linux setups), pip and pip3 can point at a different Python than python3. If installs look wrong, install through the interpreter you run:

python3 -m pip install kintic-sdk

If pip keeps reinstalling an older wheel after a new PyPI release, bypass the HTTP cache once:

python3 -m pip install --upgrade --no-cache-dir kintic-sdk

pip install kintic-sdk always installs the latest release on PyPI (no version pin needed). Pin only if you want reproducible builds, e.g. kintic-sdk==0.1.6 after you verify that release.

Quick start (3 lines)

import os
import kintic

tracer = kintic.init(api_key=os.environ["KINTIC_API_KEY"])

@tracer.decision(agent="support-bot", policy="refund_v1.2")
def handle_request(*, user_message: str) -> str:
    return "ok"

Point base_url at your API (default https://kintic.dev; SDK posts to /api/ingest, and direct integrations can also use /api/v1/ingest).

Not using Python? Any runtime can POST JSON to https://kintic.dev/api/v1/ingest with header x-api-key: kintic_live_<keyId>.<secret>. See Integration everywhere for cURL, Node/TypeScript fetch, Rust (reqwest), VS Code extensions, and reliability tips.

tracer = kintic.init(
    api_key=os.environ["KINTIC_API_KEY"],
    base_url="http://localhost:4000",
    debug=True,
)

Belief, policy, and cost

@tracer.decision(agent="refunds", policy={"version": "refund_policy_v1.2"})
def assess(*, order: dict, belief_state: dict) -> dict:
    return {
        "eligible": True,
        "reason": "Within window",
        "cost_impact": 49.0,
    }

Pass belief_state, available_information, and confidence in kwargs to attach structured context. Return a dict with cost_impact (or costImpact) to record dollar exposure on the decision.

How it works

• @tracer.decision(agent, policy=..., delegation_chain=...) wraps any callable.
• Ingest shipping is non-blocking: events are queued in-memory and shipped by a background worker.
• Queue and delivery behavior are configurable (queue_max_size, request_timeout_sec, max_retries, retry_backoff_sec).
• Delivery counters are available via tracer.get_delivery_stats() (enqueued, sent_ok, send_failed, dropped_queue_full, retries_total).
• Runtime diagnostics are available via tracer.health_snapshot() (worker status, queue config, and counters).
• For graceful shutdown in long-running services/tests, call tracer.close().
• The decorator auto-captures prompt-like args/kwargs, call stack, LLM calls, and full traceback on exceptions.

Automatic context capture

With @tracer.decision, the SDK now captures richer context automatically:

• Prompt/message-like function inputs as promptContext
• LangChain object metadata (memory, prompt template, model name, recent messages)
• LLM calls made during execution (OpenAI, Anthropic, LangChain invoke paths): model, prompt, response, latency
• Python call stack at decision time
• Exception details (failed, exceptionType, full traceback) if the function raises

Integration examples

Example 1 — Zero instrumentation with patch()

import kintic
import anthropic

tracer = kintic.init(api_key="kintic_live_<keyId>.<secret>")
kintic.patch(agent="my-agent", policy="v1.0")

# Now all anthropic calls are automatically captured
client = anthropic.Anthropic()
response = client.messages.create(...)  # Kintic captures this automatically

Pass providers to patch only the SDKs you use (install optional extras as needed):

kintic.patch(
    agent="my-agent",
    policy="v1.0",
    providers=[
        "openai",
        "anthropic",
        "gemini",   # google-genai and/or google-generativeai
        "groq",
        "mistral",
        "cohere",
    ],
)

Provider	Extra	Patched entrypoint
OpenAI	(base SDK)	chat.completions.create
Anthropic	(base SDK)	messages.create
Gemini	kintic-sdk[gemini] or [gemini-legacy]	google.genai generate_content / legacy GenerativeModel.generate_content
Groq	kintic-sdk[groq]	OpenAI-compatible chat.completions.create
Mistral	kintic-sdk[mistral]	Chat.complete
Cohere	kintic-sdk[cohere]	ClientV2.chat and legacy Client.chat

pip install "kintic-sdk[gemini,groq,mistral,cohere]"
# or: pip install "kintic-sdk[all]"

Example 2 — LangChain integration

from kintic.integrations.langchain import KinticCallbackHandler
import kintic

tracer = kintic.init(api_key="kintic_live_<keyId>.<secret>")
handler = KinticCallbackHandler(tracer=tracer, agent="my-langchain-agent", policy="v1.0")

llm = ChatOpenAI(callbacks=[handler])
agent = initialize_agent(tools, llm, callbacks=[handler])
result = agent.run("Process this refund")
# Rich reasoning + intermediate-step context (LLM/tool/agent callbacks) is captured

Example 3 — AutoGen (agentchat teams)

Requires pip install "kintic-sdk[autogen]" and autogen-agentchat.

import asyncio
import kintic
from kintic import run_observed_team

tracer = kintic.init(api_key="kintic_live_<keyId>.<secret>")

async def main():
    # team = RoundRobinGroupChat(...)  # your autogen-agentchat team
    result = await run_observed_team(
        team,
        tracer=tracer,
        task="Approve refund for order ORD-8821",
        agent="refund-autogen-team",
        policy="refund_v1.2",
        echo=True,
    )
    print(result.stop_reason)

asyncio.run(main())

Legacy AutoGen / AG2 (ConversableAgent): use LegacyAutoGenCollector, attach_legacy_observers, then collector.finish(outcome=...) after initiate_chat. See kintic.integrations.autogen.

Example 4 — CrewAI

Requires pip install "kintic-sdk[crewai]".

import kintic
from crewai import Agent, Crew, Task

tracer = kintic.init(api_key="kintic_live_<keyId>.<secret>")

# Keep listener in scope (module level) — registers on CrewAI event bus
kintic_listener = kintic.install_crewai_listener(
    tracer,
    agent="refund-crew",
    policy="refund_v1.2",
)

crew = Crew(agents=[...], tasks=[...])
result = kintic.run_observed_crew(
    crew,
    tracer=tracer,
    agent="refund-crew",
    policy="refund_v1.2",
    inputs={"order_id": "ORD-8821"},
)

Example 5 — LlamaIndex

Requires pip install "kintic-sdk[llamaindex]".

import kintic

tracer = kintic.init(api_key="kintic_live_<keyId>.<secret>")
kintic.attach_llamaindex_settings(tracer, agent="rag-agent", policy="v1")

# Or per query:
response = kintic.run_observed_query(
    query_engine,
    "What is our refund policy for order ORD-8821?",
    tracer=tracer,
    agent="rag-agent",
    policy="v1",
)

Example 6 — LangGraph

Requires pip install "kintic-sdk[langgraph]".

import kintic

tracer = kintic.init(api_key="kintic_live_<keyId>.<secret>")

# One decision when the graph finishes:
result = kintic.run_observed_invoke(
    compiled_graph,
    {"messages": [...]},
    tracer=tracer,
    agent="support-graph",
    policy="v1",
)

# Or one decision per node (stream_mode="updates"):
for chunk in kintic.run_observed_stream(
    compiled_graph,
    {"messages": [...]},
    tracer=tracer,
    agent="support-graph",
    policy="v1",
):
    ...

Per-node callback mode: create_langgraph_handler(..., ship_mode="per_node") and pass via merge_langgraph_config.

Example 7 — Google ADK

Requires pip install "kintic-sdk[google-adk]".

from google.adk.agents import LlmAgent
import kintic

tracer = kintic.init(api_key="kintic_live_<keyId>.<secret>")
adk_cb = kintic.install_adk_callbacks(tracer, agent="support-adk", policy="v1")

agent = LlmAgent(
    name="support-adk",
    model="gemini-2.0-flash",
    instruction="Be helpful.",
    **adk_cb.as_agent_kwargs(),
)
# Each agent turn ships one Kintic decision on after_agent_callback

Example 8 — Manual decorator (existing)

@tracer.decision(agent="my-agent", policy="v1.0")
def my_function(context):
    return decide(context)

Links

• Product & docs: https://kintic.dev
• Generate an API key: Kintic dashboard → Settings → API Keys (kintic_live_<keyId>.<secret>).
• Keys are shown once. Store them securely.
• Legacy kintic_live_<keyId> keys are temporarily supported for migration; rotate to the new format.

Requirements

• Python 3.8+
• requests >= 2.28.0