AgentPulse Documentation

Product analytics for AI-powered conversational products. Track sessions, not traces.

Installation

terminal
pip install agentpulse # With auto-instrumentation for specific providers: pip install agentpulse[openai] pip install agentpulse[anthropic] pip install agentpulse[langchain] pip install agentpulse[all]

Requires Python 3.11+. The only required dependency is httpx. Provider integrations (OpenAI, Anthropic, LangChain) are optional extras.

Quick Start

Add AgentPulse to your AI bot in 5 lines:

your_bot.py
import agentpulse agentpulse.init("ap_prod_...", product="my-bot") def handle_message(user_id, text): session = agentpulse.session(user_id) session.message(role="user", content=text) # ... your agent logic ... session.llm_call(response) # auto-extracts model, tokens, cost session.message(role="agent", content=reply) session.outcome("artifact") # value delivered session.payment(amount=29.00, currency="BRL")

After init, calls are batched and sent in the background every 5 seconds. The SDK never throws — all transport errors are silently logged.

Core Concepts

Session

A session is one complete user conversation with your AI agent. It's the primary unit of analysis. Sessions are keyed by user_id and auto-timeout after 30 minutes of inactivity.

Event

Something that happened within a session. Types: message.user, message.agent, action.llm_call, action.tool_call, action.cache_hit, outcome.artifact, outcome.payment, outcome.handoff, outcome.abandon, feedback.explicit, feedback.implicit.

Funnel

Configurable conversion stages that every session progresses through. Default: Reached → Engaged → Artifact → Converted.

Margin

revenue - total_cost per session. The unit economics of one conversation. This is what AgentPulse is built to answer.

agentpulse.init()

Initialize the SDK. Call once at application startup.

python
agentpulse.init( api_key="ap_prod_...", # Required - your product API key product="my-bot", # Required - product identifier base_url="https://agents-pulse.com", funnel_stages=["reached", "engaged", "artifact", "converted"], session_timeout_minutes=30, flush_interval_seconds=5.0, max_queue_size=10_000, track_content=False, debug=False, )
ParameterTypeDefaultDescription
api_keystrrequiredProduct API key (ap_prod_...)
productstrrequiredProduct identifier
base_urlstrhttps://agents-pulse.comAgentPulse server URL
funnel_stageslist[str]4 stagesCustom funnel stage names
session_timeout_minutesint30Inactivity timeout before auto-close
flush_interval_secondsfloat5.0How often to batch-send events
max_queue_sizeint10,000Max events in memory queue
track_contentboolFalseSend message text (privacy-sensitive)
debugboolFalseEnable debug logging

agentpulse.session()

Get or create a session for a user. If a session exists and hasn't timed out, returns it. Otherwise creates a new one.

python
session = agentpulse.session( user_id="5511999887766", metadata={"channel": "whatsapp"} )

Session API

session.message()

Track a user or agent message.

python
session.message(role="user", content="I need a divorce petition") session.message(role="agent", content=reply_text)

session.llm_call()

Track an LLM API call. Pass the raw response object for auto-extraction, or provide fields manually.

python
# Auto-extract from OpenAI / Anthropic response session.llm_call(response) # Manual session.llm_call( model="gpt-4o", input_tokens=100, output_tokens=50, duration_ms=1200, )

session.tool_call()

Track a tool/API call (search, database, external service).

python
session.tool_call("tavily_search", duration_ms=150, cost_usd=0.002)

session.cache_hit()

Track a cache hit that saved an LLM call.

python
session.cache_hit("semantic_cache", saved_cost_usd=0.01)

session.outcome()

Record a business outcome for this session.

python
session.outcome("artifact", metadata={"type": "petition"}) session.outcome("handoff") # escalated to human session.outcome("error")

session.payment()

Track a payment event.

python
session.payment(amount=29.00, currency="BRL")

session.feedback()

Track explicit or implicit user feedback.

python
session.feedback("explicit", "thumbs_up") session.feedback("implicit", "download")

session.end()

Manually end a session. Optional — sessions auto-end on timeout.

python
session.end("converted")

Session Properties

PropertyTypeDescription
session.conversation_depthintNumber of user messages
session.total_costfloatSum of all LLM/tool costs (USD)
session.revenuefloatSum of all payments (USD)
session.marginfloatrevenue - total_cost
session.is_endedboolWhether session is closed

agentpulse.instrument()

Wrap an LLM client for automatic call tracking. The wrapped client behaves identically — it just records every call.

python
from openai import OpenAI client = agentpulse.instrument(OpenAI()) # All calls now auto-tracked to the active session response = client.chat.completions.create( model="gpt-4o", messages=[{"role": "user", "content": text}] )

agentpulse.shutdown()

Flush pending events and stop background threads. Call before your app exits.

python
agentpulse.shutdown()

OpenAI Integration

terminal
pip install agentpulse[openai]
python
import agentpulse from openai import OpenAI agentpulse.init(api_key="ap_prod_...", product="my-bot") client = agentpulse.instrument(OpenAI()) session = agentpulse.session("user123") response = client.chat.completions.create( model="gpt-4o", messages=[{"role": "user", "content": "Hello"}] ) # Cost, tokens, model, and duration automatically captured

Anthropic Integration

terminal
pip install agentpulse[anthropic]
python
import agentpulse from anthropic import Anthropic agentpulse.init(api_key="ap_prod_...", product="my-bot") client = agentpulse.instrument(Anthropic()) session = agentpulse.session("user123") response = client.messages.create( model="claude-sonnet-4-20250514", max_tokens=1024, messages=[{"role": "user", "content": "Hello"}] )

LangChain Integration

terminal
pip install agentpulse[langchain]
python
from agentpulse.instruments.langchain import AgentPulseCallbackHandler session = agentpulse.session("user123") handler = AgentPulseCallbackHandler(session) chain.invoke({"input": "..."}, config={"callbacks": [handler]})

API Authentication

All API requests require an X-API-Key header. There are two key types:

Key TypePrefixScopeUsed By
Product Keyap_prod_Single productPython SDK
Dashboard Keyap_dash_All products in orgDashboard UI
Note: Product keys are scoped to a single product and used by the SDK. Dashboard keys see all products in your organization and are used for the web dashboard login.

POST /v1/events

Batch ingest events from the SDK. Typically you don't call this directly — the SDK handles it.

curl
curl -X POST https://agents-pulse.com/v1/events \ -H "X-API-Key: ap_prod_..." \ -H "Content-Type: application/json" \ -d '{ "events": [{ "session_id": "sess_abc123", "event_id": "evt_1", "event_type": "message.user", "timestamp": "2026-03-06T12:00:00Z", "user_id": "user_1", "product_id": "my-bot" }] }'

GET /v1/funnel

Returns conversion funnel data.

ParamTypeDescription
fromdate/datetimeStart of period (RFC3339 or YYYY-MM-DD)
todate/datetimeEnd of period
topicstringFilter by conversation topic
product_idstringRequired for dashboard keys

GET /v1/sessions

List sessions with filtering.

ParamTypeDescription
topicstringFilter by topic
outcomestringFilter by outcome
limitintPage size (default 50)
offsetintPagination offset
product_idstringRequired for dashboard keys

Get a single session with all events: GET /v1/sessions/{id}

GET /v1/economics

Returns unit economics: total revenue, total cost, gross margin, averages.

ParamTypeDescription
fromdate/datetimeStart of period
todate/datetimeEnd of period
product_idstringRequired for dashboard keys

GET /v1/topics

Returns topic breakdown with session counts, conversion rates, and revenue.

ParamTypeDescription
fromdate/datetimeStart of period
todate/datetimeEnd of period
limitintMax topics returned (default 20)
product_idstringRequired for dashboard keys

Configuration Options

The SDK reads configuration from the init() call. For local development, use debug mode:

python
agentpulse.init( api_key="console", # prints events to stdout product="test", debug=True, )

Privacy

By default, message content is NOT sent to AgentPulse servers. Only metadata is tracked:

To enable content tracking: agentpulse.init(..., track_content=True)

Multiple Environments

Use separate products for dev, staging, and production. Each product gets its own API key and isolated data.

python
import os agentpulse.init( api_key=os.environ["AGENTPULSE_API_KEY"], # different per env product=os.environ.get("AGENTPULSE_PRODUCT", "my-bot"), )
Tip: Create separate products per environment (e.g., "MyBot Dev", "MyBot Staging", "MyBot Production") under the same organization. Use a dashboard key to view all of them in the web UI with the product switcher.