No project description provided
Project description
Paid is the all-in-one, drop-in Business Engine for AI Agents that handles your pricing, subscriptions, margins, billing, and renewals with just 5 lines of code. The Paid Python library provides convenient access to the Paid API from Python applications.
Documentation
See the full API docs here
Installation
You can install the package using pip:
pip install paid-python
Usage
The client needs to be configured with your account's API key, which is available in the Paid dashboard.
from paid import Paid
client = Paid(token="API_KEY")
client.customers.create_a_new_customer(
name="name"
)
Request And Response Types
The SDK provides Python classes for all request and response types. These are automatically handled when making API calls.
# Example of creating a customer
response = client.customers.create_a_new_customer(
name="John Doe",
)
# Access response data
print(response.name)
print(response.email)
Exception Handling
When the API returns a non-success status code (4xx or 5xx response), the SDK will raise an appropriate error.
from paid import BadRequestError, NotFoundError
from paid.core.api_error import ApiError
try:
client.customers.create_a_new_customer(name="John Doe")
except BadRequestError as e:
print(e.status_code) # 400
print(e.body) # ErrorResponse with error details
except NotFoundError as e:
print(e.status_code) # 404
print(e.body)
except ApiError as e:
# Catch-all for other API errors
print(e.status_code)
print(e.body)
Logging
Supported log levels are DEBUG, INFO, WARNING, ERROR, and CRITICAL.
For example, to set the log level to debug, you can set the environment variable:
export PAID_LOG_LEVEL=DEBUG
Defaults to ERROR.
Environment Variables
The Paid SDK supports the following environment variables for configuration:
PAID_API_KEY
Your Paid API key for authentication. This is used as a fallback when you don't explicitly pass the token parameter to the Paid() client or initialize_tracing().
export PAID_API_KEY="your_api_key_here"
You can then initialize the client without passing the token:
from paid import Paid
# API key is read from PAID_API_KEY environment variable
client = Paid()
PAID_ENABLED
Controls whether Paid tracing is enabled. Set to false (case-insensitive) to disable all tracing functionality.
export PAID_ENABLED=false
This is useful for:
- Development/testing environments where tracing isn't needed
- Temporarily disabling tracing without modifying code
- Feature flagging in different deployment environments
Defaults to true if not set.
PAID_LOG_LEVEL
Sets the logging level for Paid SDK operations. See the Logging section for details.
PAID_OTEL_COLLECTOR_ENDPOINT
Overrides the default OpenTelemetry collector endpoint URL. Only needed if you want to route traces to a custom endpoint.
export PAID_OTEL_COLLECTOR_ENDPOINT="https://your-custom-endpoint.com:4318/v1/traces"
Defaults to https://collector.agentpaid.io:4318/v1/traces.
Cost Tracking via OTEL tracing
Simple Decorator and Context Manager Methods
The easiest way to add cost tracking is using the @paid_tracing decorator or context manager:
As a Decorator
from paid.tracing import paid_tracing
@paid_tracing("<external_customer_id>", external_product_id="<optional_external_product_id>")
def some_agent_workflow(): # your function
# Your logic - use any AI providers with Paid wrappers or send signals with signal().
# This function is typically an event processor that should lead to AI calls or events emitted as Paid signals
As a Context Manager
You can also use paid_tracing as a context manager with with statements:
from paid.tracing import paid_tracing
# Synchronous
with paid_tracing("customer_123", external_product_id="product_456"):
result = workflow()
# Asynchronous
async with paid_tracing("customer_123", external_product_id="product_456"):
result = await workflow()
Both approaches:
- Initialize tracing using your API key you provided to the Paid client, falls back to
PAID_API_KEYenvironment variable. - Handle both sync and async functions/code blocks
- Gracefully fall back to normal execution if tracing fails
- Support the same parameters:
external_customer_id,external_product_id,tracing_token,store_prompt,metadata
- Note - if it happens that you're calling
paid_tracingfrom non-main thread, then it's advised to initialize from main thread:
from paid.tracing import initialize_tracing
initialize_tracing()
initialize_tracingalso accepts optional arguments like OTEL collector endpoint and api key if you want to reroute your tracing somewhere else :)
Using the Paid wrappers
You can track usage costs by using Paid wrappers around your AI provider's SDK. As of now, the following SDKs' APIs are wrapped:
openai
openai-agents (as a hook)
anthropic
langchain (as a hook)
llamaindex
bedrock (boto3)
mistral
gemini (google-genai)
Example usage:
from openai import OpenAI
from paid.tracing import paid_tracing, initialize_tracing
from paid.tracing.wrappers.openai import PaidOpenAI
initialize_tracing()
openAIClient = PaidOpenAI(OpenAI(
# This is the default and can be omitted
api_key="<OPENAI_API_KEY>",
))
@paid_tracing("your_external_customer_id", external_product_id="your_external_product_id")
def image_generate():
response = openAIClient.images.generate(
model="dall-e-3",
prompt="A sunset over mountains",
size="1024x1024",
quality="hd",
style="vivid",
n=1
)
return response
image_generate()
Passing User Metadata
You can attach custom metadata to your traces by passing a metadata dictionary to the paid_tracing() decorator or context manager. This metadata will be stored with the trace and can be used to filter and query traces later.
initialize_tracing()
openai_client = PaidOpenAI(OpenAI(api_key="<OPENAI_API_KEY>"))
@paid_tracing(
"customer_123",
external_product_id="product_123",
metadata={
"campaign_id": "campaign_456",
"environment": "production",
"user_tier": "enterprise"
}
)
def process_event(event):
"""Process event with custom metadata"""
response = openai_client.chat.completions.create(
model="gpt-4",
messages=[{"role": "user", "content": event.content}]
)
signal("event_processed", enable_cost_tracing=True)
return response
process_event(incoming_event)
```
```python
from paid.tracing import paid_tracing, signal, initialize_tracing
from paid.tracing.wrappers import PaidOpenAI
from openai import OpenAI
initialize_tracing()
openai_client = PaidOpenAI(OpenAI(api_key="<OPENAI_API_KEY>"))
def process_event(event):
"""Process event with custom metadata"""
response = openai_client.chat.completions.create(
model="gpt-4",
messages=[{"role": "user", "content": event.content}]
)
signal("event_processed", enable_cost_tracing=True)
return response
# Pass metadata to context manager
with paid_tracing(
"customer_123",
external_product_id="product_123",
metadata={
"campaign_id": "campaign_456",
"environment": "production",
"user_tier": "enterprise"
}
):
process_event(incoming_event)
```
```typescript
// Metadata support is not yet available in the Node.js SDK.
// Please use Python for passing custom metadata to traces.
```
Querying Traces by Metadata
Once you've added metadata to your traces, you can filter traces using the metadata parameter in the traces API endpoint:
# Filter by single metadata field
curl -G "https://api.paid.ai/api/organizations/{orgId}/traces" \
--data-urlencode 'metadata={"campaign_id":"campaign_456"}' \
-H "Authorization: Bearer YOUR_API_KEY"
# Filter by multiple metadata fields (all must match)
curl -G "https://api.paid.ai/api/organizations/{orgId}/traces" \
--data-urlencode 'metadata={"campaign_id":"campaign_456","environment":"production"}' \
-H "Authorization: Bearer YOUR_API_KEY"
Auto-Instrumentation (OpenTelemetry Instrumentors)
For maximum convenience, you can use OpenTelemetry auto-instrumentation to automatically track costs without modifying your AI library calls. This approach uses official OpenTelemetry instrumentors for supported AI libraries.
Quick Start
from paid import Paid
from paid.tracing import paid_autoinstrument, initialize_tracing
from openai import OpenAI
# Initialize Paid SDK
client = Paid(token="PAID_API_KEY")
initialize_tracing()
paid_autoinstrument() # instruments all available: anthropic, gemini, openai, openai-agents, bedrock, langchain
# Now all OpenAI calls will be automatically traced
openai_client = OpenAI(api_key="<OPENAI_API_KEY>")
@paid_tracing("your_external_customer_id", external_product_id="your_external_product_id")
def chat_with_gpt():
response = openai_client.chat.completions.create(
model="gpt-4",
messages=[{"role": "user", "content": "Hello!"}]
)
return response
chat_with_gpt() # Costs are automatically tracked!
Supported Libraries
Auto-instrumentation supports the following AI libraries:
anthropic - Anthropic SDK
gemini - Google Generative AI (google-generativeai)
openai - OpenAI Python SDK
openai-agents - OpenAI Agents SDK
bedrock - AWS Bedrock (boto3)
langchain - LangChain framework
Selective Instrumentation
If you only want to instrument specific libraries, pass them to paid_autoinstrument():
from paid.tracing import paid_autoinstrument
# Instrument only Anthropic and OpenAI
paid_autoinstrument(libraries=["anthropic", "openai"])
How It Works
- Auto-instrumentation uses official OpenTelemetry instrumentors for each AI library
- It automatically wraps library calls without requiring you to use Paid wrapper classes
- Works seamlessly with
@paid_tracing()decorator or context manager - Costs are tracked in the same way as when using manual wrappers
- Should be called once during application startup, typically before creating AI client instances
Signaling via OTEL tracing
Signals allow you to emit events within your tracing context. They have access to all tracing information, so you need fewer arguments compared to manual API calls.
Use the signal() function which must be called within an active @paid_tracing() context (decorator or context manager).
Here's an example of how to use it:
from paid.tracing import paid_tracing, signal
@paid_tracing("your_external_customer_id", external_product_id="your_external_product_id")
def do_work():
# ...do some work...
signal(
event_name="<your_signal_name>",
data={ } # optional data (ex. manual cost tracking data)
)
do_work()
Same approach with context manager:
from paid.tracing import paid_tracing, signal
def do_work():
# ...do some work...
signal(
event_name="<your_signal_name>",
data={ } # optional data (ex. manual cost tracking data)
)
# Use context manager instead
with paid_tracing("your_external_customer_id", external_product_id="your_external_product_id"):
do_work()
Signal-costs - Attaching cost traces to a signal
If you want a signal to carry information about costs, then the signal should be sent from the same tracing context as the wrappers and hooks that recorded those costs.
This will look something like this:
from paid.tracing import paid_tracing, signal
@paid_tracing("your_external_customer_id", external_product_id="your_external_product_id")
def do_work():
# ... your workflow logic
# ... your AI calls made through Paid wrappers or hooks
signal(
event_name="<your_signal_name>",
data={ }, # optional data (ex. manual cost tracking data)
enable_cost_tracing=True, # set this flag to associate it with costs
)
# ... your workflow logic
# ... your AI calls made through Paid wrappers or hooks (can be sent after the signal too)
do_work()
Then, all of the costs traced in @paid_tracing() context are related to that signal.
Signal-costs - Distributed tracing
Sometimes your agent workflow cannot fit into a single traceable function like above, because it has to be disjoint for whatever reason. It could even be running across different machines.
For such cases, you can pass a tracing token directly to @paid_tracing() or context manager to link distributed traces together.
Using tracing_token parameter (Recommended)
The simplest way to implement distributed tracing is to pass the token directly to the decorator or context manager:
from paid.tracing import paid_tracing, signal, generate_tracing_token, initialize_tracing
from paid.tracing.wrappers.openai import PaidOpenAI
from openai import OpenAI
initialize_tracing()
openai_client = PaidOpenAI(OpenAI(api_key="<OPENAI_API_KEY>"))
# Process 1: Generate token and do initial work
token = generate_tracing_token()
print(f"Tracing token: {token}")
# Store token for other processes (e.g., in Redis, database, message queue)
save_to_storage("workflow_123", token)
@paid_tracing("customer_123", tracing_token=token, external_product_id="product_123")
def process_part_1():
# AI calls here will be traced
response = openai_client.chat.completions.create(
model="gpt-4",
messages=[{"role": "user", "content": "Analyze data"}]
)
# Signal without cost tracing
signal("part_1_complete", enable_cost_tracing=False)
process_part_1()
# Process 2 (different machine/process): Retrieve and use token
token = load_from_storage("workflow_123")
@paid_tracing("customer_123", tracing_token=token, external_product_id="product_123")
def process_part_2():
# AI calls here will be linked to the same trace
response = openai_client.chat.completions.create(
model="gpt-4",
messages=[{"role": "user", "content": "Generate response"}]
)
# Signal WITH cost tracing - links all costs from both processes
signal("workflow_complete", enable_cost_tracing=True)
process_part_2()
# No cleanup needed - token is scoped to the decorated function
Using context manager instead of decorator:
from paid.tracing import paid_tracing, signal, generate_tracing_token, initialize_tracing
from paid.tracing.wrappers.openai import PaidOpenAI
from openai import OpenAI
initialize_tracing()
# Initialize
openai_client = PaidOpenAI(OpenAI(api_key="<OPENAI_API_KEY>"))
# Process 1: Generate token and do initial work
token = generate_tracing_token()
save_to_storage("workflow_123", token)
with paid_tracing("customer_123", external_product_id="product_123", tracing_token=token):
response = openai_client.chat.completions.create(
model="gpt-4",
messages=[{"role": "user", "content": "Analyze data"}]
)
signal("part_1_complete", enable_cost_tracing=False)
# Process 2: Retrieve and use the same token
token = load_from_storage("workflow_123")
with paid_tracing("customer_123", external_product_id="product_123", tracing_token=token):
response = openai_client.chat.completions.create(
model="gpt-4",
messages=[{"role": "user", "content": "Generate response"}]
)
signal("workflow_complete", enable_cost_tracing=True)
Manual Cost Tracking
If you would prefer to not use Paid to track your costs automatically but you want to send us the costs yourself, then you can use manual cost tracking mechanism. Just attach the cost information in the following format to a signal payload:
from paid import Paid, Signal
client = Paid(token="<PAID_API_KEY>")
signal = Signal(
event_name="<your_signal_name>",
agent_id="<your_agent_id>",
external_customer_id="<your_external_customer_id>",
data = {
"costData": {
"vendor": "<any_vendor_name>", # can be anything, traces are grouped by vendors in the UI
"cost": {
"amount": 0.002,
"currency": "USD"
},
"gen_ai.response.model": "<ai_model_name>",
}
}
)
client.usage.record_bulk(signals=[signal])
Alternatively the same costData payload can be passed to OTLP signaling mechanism:
from paid.tracing import paid_tracing, signal
@paid_tracing("your_external_customer_id", external_product_id="your_external_product_id")
def do_work():
# ...do some work...
signal(
event_name="<your_signal_name>",
data={
"costData": {
"vendor": "<any_vendor_name>", # can be anything, traces are grouped by vendors in the UI
"cost": {
"amount": 0.002,
"currency": "USD"
},
"gen_ai.response.model": "<ai_model_name>",
}
}
)
do_work()
Manual Usage Tracking
If you would prefer to send us raw usage manually (without wrappers) and have us compute the cost, you can attach usage data in the following format:
from paid import Paid, Signal
client = Paid(token="<PAID_API_KEY>")
signal = Signal(
event_name="<your_signal_name>",
agent_id="<your_agent_id>",
external_customer_id="<your_external_customer_id>",
data = {
"costData": {
"vendor": "<any_vendor_name>", # can be anything, traces are grouped by vendors in the UI
"attributes": {
"gen_ai.response.model": "gpt-4.1-mini",
"gen_ai.usage.input_tokens": 100,
"gen_ai.usage.output_tokens": 300,
"gen_ai.usage.cached_input_tokens": 600,
},
}
}
)
client.usage.record_bulk(signals=[signal])
Same but via OTEL signaling:
from paid.tracing import paid_tracing, signal
@paid_tracing("your_external_customer_id", external_product_id="your_external_product_id")
def do_work():
# ...do some work...
signal(
event_name="<your_signal_name>",
data={
"costData": {
"vendor": "<any_vendor_name>", # can be anything, traces are grouped by vendors in the UI
"attributes": {
"gen_ai.response.model": "gpt-4.1-mini",
"gen_ai.usage.input_tokens": 100,
"gen_ai.usage.output_tokens": 300,
"gen_ai.usage.cached_input_tokens": 600,
},
}
}
)
do_work()
Async Support
All of the functionality from above is available in async flavor too.
Async Client
Use AsyncPaid instead of Paid for async operations:
from paid import AsyncPaid
client = AsyncPaid(token="API_KEY")
# Async API calls
customer = await client.customers.create(name="John Doe")
Async Cost Tracking with Decorator
The @paid_tracing decorator automatically handles both sync and async functions:
from openai import AsyncOpenAI
from paid.tracing import paid_tracing, initialize_tracing
from paid.tracing.wrappers.openai import PaidAsyncOpenAI
initialize_tracing()
# Wrap the async OpenAI client
openai_client = PaidAsyncOpenAI(AsyncOpenAI(api_key="<OPENAI_API_KEY>"))
@paid_tracing("your_external_customer_id", external_product_id="your_external_product_id")
async def generate_image():
response = await openai_client.images.generate(
model="dall-e-3",
prompt="A sunset over mountains",
size="1024x1024",
quality="hd",
n=1
)
return response
# Call the async function
await generate_image()
Async Signaling
The signal() function works seamlessly in async contexts:
from paid.tracing import paid_tracing, signal, initialize_tracing
from paid.tracing.wrappers.openai import PaidAsyncOpenAI
from openai import AsyncOpenAI
initialize_tracing()
openai_client = PaidAsyncOpenAI(AsyncOpenAI(api_key="<OPENAI_API_KEY>"))
@paid_tracing("your_external_customer_id", external_product_id="your_external_product_id")
async def do_work():
# Perform async AI operations
response = await openai_client.chat.completions.create(
model="gpt-4",
messages=[{"role": "user", "content": "Hello!"}]
)
# Send signal (works in async context)
signal(
event_name="<your_signal_name>",
enable_cost_tracing=True # Associate with traced costs
)
return response
# Execute
await do_work()
Paid OTEL Tracer Provider
If you would like to use the Paid OTEL tracer provider:
from paid.tracing import get_paid_tracer_provider
paid_tracer_provider = get_paid_tracer_provider()
Contributing
While we value open-source contributions to this SDK, this library is generated programmatically. Additions made directly to this library would have to be moved over to our generation code, otherwise they would be overwritten upon the next generated release. Feel free to open a PR as a proof of concept, but know that we will not be able to merge it as-is. We suggest opening an issue first to discuss with us!
On the other hand, contributions to the README are always very welcome!
Project details
Release history Release notifications | RSS feed
Download files
Download the file for your platform. If you're not sure which to choose, learn more about installing packages.
Source Distribution
Built Distribution
Filter files by name, interpreter, ABI, and platform.
If you're not sure about the file name format, learn more about wheel file names.
Copy a direct link to the current filters
File details
Details for the file paid_python-1.0.0a1.tar.gz.
File metadata
- Download URL: paid_python-1.0.0a1.tar.gz
- Upload date:
- Size: 66.2 kB
- Tags: Source
- Uploaded using Trusted Publishing? No
- Uploaded via: poetry/1.5.1 CPython/3.9.25 Linux/6.11.0-1018-azure
File hashes
| Algorithm | Hash digest | |
|---|---|---|
| SHA256 |
03684c30a55636050a3e539275428bfdfca50f75adf655e691e7b6b38bbb92cf
|
|
| MD5 |
1a3ca2a89341e17c1a0364c3a27e122c
|
|
| BLAKE2b-256 |
b83c19b3870693e0decea91903fc4205aa11957f4005d9501dfe69295db185cf
|
File details
Details for the file paid_python-1.0.0a1-py3-none-any.whl.
File metadata
- Download URL: paid_python-1.0.0a1-py3-none-any.whl
- Upload date:
- Size: 103.3 kB
- Tags: Python 3
- Uploaded using Trusted Publishing? No
- Uploaded via: poetry/1.5.1 CPython/3.9.25 Linux/6.11.0-1018-azure
File hashes
| Algorithm | Hash digest | |
|---|---|---|
| SHA256 |
2abbf66745fe391d1b3cb7372188da12e90a28e420886be962259dc7441a4908
|
|
| MD5 |
10c530562a88e5c17df22f7addb0cde4
|
|
| BLAKE2b-256 |
fad9e263f670463420265c9e10e0fae007e5a28365965c575e0c5e764fff5567
|