Overview

This guide explains how to manually submit individual event data to Pay-i using the Python SDK's ingest.units() method. This method allows you to explicitly track AI resource usage when automatic instrumentation isn't available or sufficient. For detailed API reference information, see the Python SDK Ingest API Reference.

When to Use Manual Event Submission

Manual event submission is typically used right after an AI request completes, allowing you to capture usage metrics and apply business context. This is especially valuable for applications that need to track AI usage outside of Pay-i's automatic instrumentation.

Explicit event submission is particularly valuable for these key use cases:

Tracking Custom Resources - Submit usage data for custom resources not directly supported by Pay-i
Submitting Historical Data - Record individual past events for historical data backfill
External System Integration - Connect with your existing telemetry systems and forward data to Pay-i

Note: For submitting multiple events efficiently in batches (bulk ingestion), please contact [email protected] for information on available options.

Common Workflows

Working with the Pay-i ingest.units() method involves several common patterns for submitting individual event data. This section walks you through these workflows with practical code examples.

Note: The examples in this guide use the Python SDK's client objects (Payi and AsyncPayi), which provide a resource-based interface to the Pay-i API. For details on client initialization and configuration, see the Pay-i Client Initialization guide.

Basic Event Submission

The most common pattern is submitting a single event representing one AI interaction:

from payi import Payi

# Initialize the Pay-i client
client = Payi()  # API key will be loaded from PAYI_API_KEY environment variable

# Submit an event to Pay-i
response = client.ingest.units(
    category="system.openai",
    resource="gpt-4o-mini",
    units={
        "text": {
            "input": 156,    # Number of input tokens used in the request
            "output": 1746   # Number of output tokens generated in the response
        }
    }
    # Note: The SDK automatically sets event_timestamp to datetime.now(timezone.utc) if not specified
)

# Check tracking status and cost information
print(f"Request ID: {response.request_id}")
print(f"Event timestamp: {response.event_timestamp}")
# Access cost information
print(f"Base cost: ${response.xproxy_result.cost.total.base}")

Tracking Custom AI Resources

When you're using custom or internal AI models that aren't directly supported by Pay-i's automatic instrumentation, you can manually track their usage:

from payi import Payi
from datetime import datetime, timezone

# Initialize the Pay-i client
client = Payi()

# First, ensure your custom resource exists in Pay-i
# This step can be done once during application setup
resource_response = client.categories.resources.create(
    category="custom.my_company",
    resource="proprietary-llm-v1",
    units={
        "text": {
            "input_price": 0.0001,  # $0.0001 per input token
            "output_price": 0.0005  # $0.0005 per output token
        }
    },
    # IMPORTANT: If you plan to track historical data for this resource,
    # explicitly set a start_timestamp that predates your oldest events
    start_timestamp="2024-01-01T00:00:00Z"  # Making resource valid from Jan 1, 2024
)

# Now track usage of your custom model
response = client.ingest.units(
    category="custom.my_company",
    resource="proprietary-llm-v1",
    units={
        "text": {
            "input": 215,    # Number of input tokens used in the request
            "output": 845    # Number of output tokens generated in the response
        }
    },
    user_id="customer-123",
    request_tags=["customer-support", "documentation"]
)

Historical Data Backfilling

When you need to import historical usage data into Pay-i for retroactive analysis, you must carefully manage timestamps:

from datetime import datetime
from payi import Payi

client = Payi()

# Example: Submitting a historical event from a log file
# The timestamp comes from a parsed log entry's timestamp
historical_timestamp = "2024-01-15T14:22:36Z"  # ISO 8601 format with UTC timezone (Z)

# Note: Always use UTC time for observability systems. The "Z" suffix explicitly
# indicates UTC timezone in ISO 8601 format. If omitted, the timestamp might be
# interpreted as local time, causing inconsistencies in reporting and analysis.
#
# IMPORTANT: The resource must exist and be valid at the event_timestamp!
# Pay-i validates that the resource existed at the specified timestamp.
# You will need to set start_timestamp when creating the resource (see Custom Resources example).

response = client.ingest.units(
    category="system.openai",
    resource="gpt-4", # Use a model that existed at the historical timestamp
    units={
        "text": {
            "input": 156,    # Number of input tokens used in the request
            "output": 1746   # Number of output tokens generated in the response
        }
    },
    event_timestamp=historical_timestamp,  # Timestamp when the event actually occurred
    end_to_end_latency_ms=12450,           # Latency in milliseconds
    time_to_first_token_ms=1143            # TTFT in milliseconds
)

Resource Timestamp Validation

Pay-i enforces a validation rule that every event must reference a resource that existed at the event's timestamp. This is critically important when:

Ingesting Historical Data: Events with timestamps before a resource was created will be rejected
Using Custom Resources: Custom resources are valid from their creation time by default
Integrating External Telemetry: External data often includes historical timestamps

Resource Validity Period

Every resource in Pay-i has a validity period:

start_timestamp: When the resource becomes valid (defaults to creation time if not specified)
end_timestamp: When the resource stops being valid (optional, null means "forever")

When an event is ingested, Pay-i verifies the resource was valid at event_timestamp. If no valid resource is found for that timestamp, you'll receive a 404 error with code category_resource_not_found.

Common Error Example

Error code: 404 - {"message": "Category 'custom.my_company' and resource 'proprietary-llm-v1' and event timestamp '04/24/2025 15:52:45 +00:00' not found", 'statusCode': 404, 'xproxy_error': {'code': 'category_resource_not_found'}}

This error indicates that at timestamp '04/24/2025 15:52:45 +00:00', the specified resource didn't exist or wasn't valid yet.

Solution for Historical Data

When creating resources for historical data ingestion, always set a start_timestamp that predates your oldest events:

# When creating a resource for historical data tracking
client.categories.resources.create(
    category="custom.my_company",
    resource="proprietary-llm-v1",
    units={"text": {"input_price": 0.0001, "output_price": 0.0005}},
    # Make the resource valid from before your oldest events
    start_timestamp="2024-01-01T00:00:00Z"  # Valid from Jan 1, 2024
)

External System Integration

When you need to synchronize Pay-i with your existing telemetry or observability infrastructure, you can forward events from those systems:

from payi import Payi
from typing import List, Dict, Any

# Initialize the Pay-i client
client = Payi()

# Sample telemetry data from different external systems
external_telemetry_events = [
    {
        "event_type": "ai_completion",
        "timestamp": "2024-03-22T15:45:12.345Z",
        "provider": "openai",          # OpenAI provider maps to system.openai category
        "model": "gpt-4",
        "tokens": {
            "prompt": 312,             # Input tokens
            "completion": 945          # Output tokens
        },
        "latency_ms": 2245,
        "metadata": {
            "user": "user-456",
            "project": "sales-analytics",
            "environment": "production"
        }
    },
    {
        "event_type": "ai_completion",
        "timestamp": "2024-03-22T15:48:33.120Z",
        "provider": "anthropic",       # Anthropic provider maps to system.anthropic category
        "model": "claude-3-opus-20240229",
        "tokens": {
            "prompt": 520,             # Input tokens
            "completion": 1230         # Output tokens
        },
        "latency_ms": 3150,
        "metadata": {
            "user": "user-789",
            "project": "sales-analytics",
            "environment": "production"
        }
    },
    {
        "event_type": "ai_completion",
        "timestamp": "2024-03-22T15:52:45.890Z",
        "provider": "my_company",  # Maps to custom.my_company category
        "model": "proprietary-llm-v1",    # Using the same resource defined earlier
        "tokens": {
            "prompt": 275,             # Input tokens
            "completion": 830          # Output tokens
        },
        "latency_ms": 1890,
        "metadata": {
            "user": "user-123",
            "project": "sales-analytics",
            "environment": "staging"
        }
    }
]

# Map the external telemetry format to Pay-i's expected format
def map_external_telemetry_to_payi(event: Dict[str, Any]) -> Dict[str, Any]:
    # Map provider to appropriate category
    if event["provider"] == "openai":
        # Official OpenAI models use the system.openai category
        category = "system.openai"
    elif event["provider"] == "anthropic":
        # Official Anthropic models use the system.anthropic category
        category = "system.anthropic"
    elif event["provider"] == "my_company":
        # Map to the custom.my_company category we created earlier
        category = "custom.my_company"
    else:
        # For any other provider, we create a custom category
        # (ensure these categories exist in your Pay-i system)
        category = f"custom.{event['provider'].replace('_', '.')}"
    
    # Map the event to Pay-i's format
    return {
        "category": category,
        "resource": event["model"],
        "units": {
            "text": {
                "input": event["tokens"]["prompt"],     # Number of input tokens
                "output": event["tokens"]["completion"] # Number of output tokens
            }
        },
        "event_timestamp": event["timestamp"],
        "end_to_end_latency_ms": event["latency_ms"],
        "user_id": event["metadata"]["user"],
        "request_tags": [
            event["metadata"]["project"],
            event["metadata"]["environment"],
            "external_telemetry"
        ]
    }

# Process each telemetry event
for event in external_telemetry_events:
    # Transform the external event to Pay-i format
    mapped_event = map_external_telemetry_to_payi(event)
    
    # Submit the event to Pay-i using dictionary unpacking
    try:
        response = client.ingest.units(**mapped_event)
        print(f"Processed {event['provider']} event, base cost: ${response.xproxy_result.cost.total.base}")
    except Exception as e:
        # Handle errors, which may include category_resource_not_found
        print(f"Error processing {event['provider']} event: {e}")

NOTE: Dictionary unpacking with **mapped_params is a valid approach for parameter passing with units(). It does not mean you're passing a dictionary as a parameter; rather, it converts the dictionary key-value pairs into individual named arguments.

For example, this code:
mapped_params = {
    "category": "system.openai",
    "resource": "gpt-4o-mini",
    "units": {"text": {"input": 100, "output": 200}},
    "user_id": "user-123"
}
client.ingest.units(**mapped_params)
Is equivalent to this:
client.ingest.units(
    category="system.openai",
    resource="gpt-4o-mini",
    units={"text": {"input": 100, "output": 200}},
    user_id="user-123"
)
This pattern is often used when building parameters programmatically or mapping from external data formats.

Asynchronous Event Submission

For applications using async/await patterns, you can use the asynchronous client:

import asyncio
from payi import AsyncPayi

async_client = AsyncPayi()  # API key from environment variable

async def submit_event():
    response = await async_client.ingest.units(
        category="system.openai",
        resource="gpt-4o-mini",
        units={
            "text": {
                "input": 156,    # Number of input tokens used in the request
                "output": 1746   # Number of output tokens generated in the response
            }
        }
    )
    return response

response = asyncio.run(submit_event())

Common Mistakes to Avoid

1. Passing header parameters in `extra_headers`

# INCORRECT - Don't do this!
client.ingest.units(
    # ... other parameters
    extra_headers={
        "Request-Tags": "customer-support,documentation",  # Wrong!
        "User-ID": "user-123",                             # Wrong!
        "Limit-IDs": "monthly-budget,daily-quota",         # Wrong!
        "Use-Case-Version": "2"                            # Wrong! Must be an integer, not a string
    }
)

# CORRECT way - Pass directly as parameters
client.ingest.units(
    # ... other parameters
    request_tags=["customer-support", "documentation"],
    user_id="user-123",
    limit_ids=["monthly-budget", "daily-quota"],
    use_case_version=2  # Correct: integer, not string
)

2. Using strings for list parameters

# INCORRECT - Don't pass comma-separated strings
request_tags="customer-support,documentation"  # Wrong!

# CORRECT - Use a proper list
request_tags=["customer-support", "documentation"]

3. Not handling historical timestamps properly

# INCORRECT - Creating resource without start_timestamp, then using historical data
client.categories.resources.create(
    category="custom.my_company",
    resource="proprietary-llm-v1",
    units={"text": {"input_price": 0.0001, "output_price": 0.0005}}
)

# Later trying to ingest historical data (will fail with 404)
client.ingest.units(
    category="custom.my_company",
    resource="proprietary-llm-v1",
    event_timestamp="2024-03-01T12:00:00Z",  # Date before resource creation
    # ... other parameters
)

# CORRECT - Set start_timestamp when creating resources for historical data
client.categories.resources.create(
    category="custom.my_company",
    resource="proprietary-llm-v1",
    units={"text": {"input_price": 0.0001, "output_price": 0.0005}},
    start_timestamp="2024-01-01T00:00:00Z"  # Set to before your oldest events
)

Required Parameters

When using client.ingest.units(), the following parameters are required:

category: The category the resource belongs to (e.g., "system.openai", "system.anthropic")
resource: The specific resource used (e.g., "gpt-4o-mini", "claude-3-haiku")
units: Dictionary containing unit types with input and output values

Important Notes on Category and Resource Mapping

When using the ingest.units() method, here are some key points to remember:

Both the category and resource must exist in your Pay-i system before you can track them
The resource must have existed at the specified timestamp (for historical data)
When mapping external provider names to categories:
- Official providers map to system.{provider_name} (e.g., system.openai, system.anthropic)
- Custom providers map to custom.{provider_name} (e.g., custom.my_company)
Be consistent with naming between resource creation and usage tracking
Ensure the timestamp for historical data is within the valid range for the resource

Method Reference

Method	Description	API Endpoint
`units()`	Submits a single AI resource usage event to Pay-i	Ingest Events

API Reference

For detailed information about all available methods, parameters, and response types, see the Python SDK Ingest API Reference.