Python SDK

Comprehensive AI agent observability for Python applications. The Handit.ai Python SDK provides powerful tracking and monitoring capabilities designed specifically for AI applications, with seamless integration for popular frameworks like LangChain and OpenAI.

This guide covers everything you need to integrate and use the SDK effectively in your Python AI applications.

The SDK automatically tracks function executions, model calls, and tool usage, providing detailed insights into your AI application’s behavior with minimal setup required.

Installation

terminal


pip install -U "handit-sdk>=1.9.0"

Quick Start

app.py


from handit_service import tracker
 
# Configure the tracker
tracker.config(api_key="your-api-key")
 
# Start tracking your agent
@tracker.start_agent_tracing(key="my-agent")
async def my_agent(input_data):
    # Your agent logic here
    pass

Method Reference

Method	Purpose	Minimal Example	Step-by-Step Explanation
`@tracker.start_agent_tracing(key=…)`	Root trace for each agent request (inputs, outputs, errors, total duration)	`@tracker.start_agent_tracing(key="invoice-assistant") async def process(msg): ...`	Creates an agentLogId and stores it in a global ContextVar. Any subsequent node/call inherits this ID. Sends a start event and, upon completion or failure, sends an end event with success/error status.
`@tracker.trace_agent_node("slug")`	Internal step (model, tool, RAG…) when you control the function definition	`@tracker.trace_agent_node("vector-search") def fetch(q): ...`	Links the call to the active agentLogId. Serializes arguments and return value (or exception). Publishes a node event with duration metric.
`tracker.trace_agent_node_func(func, *args, key=…)`	Same as above but without decorating the function (useful for external code or one-off calls)	`docs = await tracker.trace_agent_node_func( fetch_docs, query, key="vector-search")`	Executes the function (sync or async). Sends node event with input/output. Keeps your stack clean (no need to modify fetch_docs).
`tracker.trace_agent_node_func_sync(…)`	Strictly synchronous version (faster by avoiding asyncio)	`result = tracker.trace_agent_node_func_sync( heavy_calc, data, key="feature-engineering")`	Identical to async version but without await; designed for CPU-bound helpers.
`tracker.track_model(model, "slug")`	Intercepts all model calls (call) and tracks them	`llm = ChatOpenAI(model="gpt-4") llm = tracker.track_model(llm, "gpt4-prod") await llm.invoke("Hello")`	Monkey-patches model.call. Before execution, publishes model_call event (prompt, params). After execution, publishes model_response event (text, tokens, etc.).
`tracker.track_tool(tool, "slug")`	Same as track_model but for non-LLM functions or classes (scrapers, translators, etc.)	`def add(a,b): return a+b add = tracker.track_tool(add, "adder") await add(2,3)`	Sends tool_call with args/kwargs. Executes original logic. Sends tool_response with result or captured exception.
`tracker.intercept_requests(requests.<method>)`	Tracks outgoing HTTP requests made with requests when the URL matches Handit’s remote list	`requests.post = tracker.intercept_requests(requests.post)`	Downloads urls_to_track from backend. If URL matches, publishes input/output of the call. Transparent to the rest of the code.
`tracker.capture_model(model_id, req, res)`	Fires a manual event; useful in batch pipelines or when receiving data via webhook	`tracker.capture_model("daily-batch", {"date":"2025-05-22"}, {"rows":1234})`	Sends an arbitrary input/output object to the /track endpoint, linking it (if it exists) to the current agentLogId.
`tracker.update_tracked_urls()` (internal)	Synchronizes “trackable” URLs for intercept_requests	Called automatically from intercept_requests	Ensures you don’t need to redeploy if you add a new API to monitor tomorrow.
`tracker.fetch_optimized_prompt(model_id)`	Doesn’t track, but reads performance data: returns the winning prompt for the specified model/node	`best = tracker.fetch_optimized_prompt("gpt4-prod") print(best["prompt"])`	Queries /performance/model/{id}/optimized-prompt. Use it for continuous A/B testing or self-optimization.

Core Methods Documentation

Configuration

`config(api_key: str, tracking_url: Optional[str] = None)`

Initializes the tracker with your API key and optional custom tracking URL.

tracker.py


def config(self, api_key: str, tracking_url: Optional[str] = None):
    """
    Configure the tracker with API key and optional custom tracking URL.
    
    Args:
        api_key (str): Your Handit.ai API key
        tracking_url (str, optional): Custom tracking server URL
        
    Raises:
        ValueError: If API key is not provided
    """

What it does:

Sets authentication credentials for all tracking operations
Optionally configures custom tracking server endpoint
Must be called before using any other tracking functions

Agent-Level Tracing

`start_agent_tracing(key: str)`

Decorator for tracing complete agent execution workflows.

agent.py


def start_agent_tracing(self, key: str):
    """
    Decorator to trace full agent execution.
    Supports both async and sync functions.
    
    Args:
        key (str): Unique identifier for the agent
    """

Key capabilities:

Creates and manages execution context using ContextVar
Automatically handles both async and sync functions
Tracks start/end events with timing and error information
Links all nested operations to the same agent execution

Function-Level Tracing

`trace_agent_node(agent_node_id: str)`

Decorator for tracing individual functions within an agent workflow.

node.py


def trace_agent_node(self, agent_node_id: str):
    """
    Decorator to trace individual agent node execution.
    
    Args:
        agent_node_id (str): Unique identifier for the node
    """

Key capabilities:

Captures input parameters and return values
Handles both async and sync functions automatically
Reports errors with full context and stack traces
Links to parent agent execution context

`trace_agent_node_func(func, *args, key=None, **kwargs)`

Programmatic function tracing without decorators.

function.py


async def trace_agent_node_func(self, func: Callable, *args, key: str = None, **kwargs):
    """
    Trace any function call without modifying the function definition.
    
    Args:
        func (Callable): The function to trace
        *args: Positional arguments for the function
        key (str, optional): Tracking identifier (defaults to function name)
        **kwargs: Keyword arguments for the function
    """

Key capabilities:

Works with both sync and async functions
No need to modify original function definitions
Captures complete input/output data
Handles errors gracefully with full context

`trace_agent_node_func_sync(func, *args, key=None, **kwargs)`

Optimized synchronous version for CPU-bound operations.

function_sync.py


def trace_agent_node_func_sync(self, func: Callable, *args, key: str = None, **kwargs):
    """
    Synchronous version optimized for CPU-bound operations.
    """

Model & Tool Tracking

`track_model(model, model_id: str)`

Creates a tracked version of AI models (LLMs, embeddings, etc.).

model.py


def track_model(self, model, model_id: str):
    """
    Creates a tracked version of an AI model.
    
    Args:
        model: The model instance to track
        model_id (str): Unique identifier for tracking
    """

Key capabilities:

Automatically intercepts all model calls
Captures prompts, responses, and performance metrics
Works with LangChain, OpenAI, and custom models
Maintains original model functionality

`track_tool(tool, tool_id: str)`

Creates a tracked version of any tool or function.

tool.py


def track_tool(self, tool, tool_id: str):
    """
    Creates a tracked version of a tool or function.
    
    Args:
        tool: The tool/function to track
        tool_id (str): Unique identifier for tracking
    """

Key capabilities:

Works with any callable object
Tracks input/output for each execution
Handles both sync and async operations
Preserves original functionality

HTTP Request Tracking

`intercept_requests(func)`

Automatically tracks HTTP requests to specified URLs.

requests.py


def intercept_requests(self, func):
    """
    Intercepts and tracks HTTP requests.
    
    Args:
        func: The request function to intercept
    """

Key capabilities:

Automatically tracks requests to configured URLs
Updates tracking list from server dynamically
Captures request/response data transparently
Works with requests library methods

Custom Event Tracking

`capture_model(model_id: str, request_body: Any, response_body: Any)`

Manually track custom events and operations.

events.py


def capture_model(self, model_id: str, request_body: Any, response_body: Any):
    """
    Manually track custom events.
    
    Args:
        model_id (str): Event identifier
        request_body (Any): Input data
        response_body (Any): Output data
    """

Key capabilities:

Manual event tracking for custom scenarios
Flexible input/output data format
Links to current agent execution context
Useful for batch operations and webhooks

Performance Optimization

`fetch_optimized_prompt(model_id: str)`

Retrieves optimized prompts based on performance data.

optimization.py


def fetch_optimized_prompt(self, model_id: str):
    """
    Fetches the best-performing prompt for a model.
    
    Args:
        model_id (str): Model identifier
        
    Returns:
        dict: Optimized prompt data
    """

Key capabilities:

Retrieves best-performing prompts from A/B testing
Enables continuous optimization workflows
Supports self-improving AI systems
Requires proper API key configuration

All tracking methods automatically handle errors and maintain execution context, ensuring reliable operation even when exceptions occur in your application code.

⚠️

Always configure the tracker with tracker.config(api_key="your-key") before using any tracking functions.

Real-World Use Cases

1. Invoice Processing Agent

invoice_agent.py


from handit_service import tracker
from langchain.chat_models import ChatOpenAI
from langchain.embeddings import OpenAIEmbeddings
import json
 
# Configure the tracker with your API key
tracker.config(api_key="your-api-key")
 
# Initialize and track models
llm = tracker.track_model(
    ChatOpenAI(model="gpt-4"),
    model_id="invoice-gpt4"
)
embeddings = tracker.track_model(
    OpenAIEmbeddings(),
    model_id="invoice-embeddings"
)
 
# Main agent function with tracing
@tracker.start_agent_tracing(key="invoice-processor")
async def process_invoice(invoice_data: dict) -> dict:
    """
    Process an invoice document and extract key information.
    
    Args:
        invoice_data (dict): Raw invoice data including text and metadata
        
    Returns:
        dict: Structured invoice information
    """
    try:
        # Extract text from invoice
        text = await extract_text(invoice_data)
        
        # Search for relevant information
        info = await search_invoice_info(text)
        
        # Generate structured output
        result = await generate_invoice_output(info)
        
        return result
    except Exception as e:
        # Errors are automatically tracked
        raise
 
# Track individual processing steps
@tracker.trace_agent_node("text-extractor")
async def extract_text(data: dict) -> str:
    """
    Extract text content from invoice data.
    
    Args:
        data (dict): Raw invoice data
        
    Returns:
        str: Extracted text content
    """
    # Implementation details...
    return text
 
@tracker.trace_agent_node("info-searcher")
async def search_invoice_info(text: str) -> dict:
    """
    Search for relevant information in invoice text.
    
    Args:
        text (str): Invoice text content
        
    Returns:
        dict: Found information
    """
    # Implementation details...
    return info
 
@tracker.trace_agent_node("output-generator")
async def generate_invoice_output(info: dict) -> dict:
    """
    Generate structured output from found information.
    
    Args:
        info (dict): Found invoice information
        
    Returns:
        dict: Structured output
    """
    # Implementation details...
    return output

2. Customer Support Chatbot

support_bot.py


from handit_service import tracker
from langchain.chat_models import ChatOpenAI
from langchain.vectorstores import Pinecone
import pinecone
 
# Configure the tracker
tracker.config(api_key="your-api-key")
 
# Initialize and track the chat model
chat_model = tracker.track_model(
    ChatOpenAI(model="gpt-4"),
    model_id="support-gpt4"
)
 
# Track the vector store
@tracker.trace_agent_node("knowledge-base")
async def search_knowledge_base(query: str) -> list:
    """
    Search the knowledge base for relevant information.
    
    Args:
        query (str): User's question
        
    Returns:
        list: Relevant documents
    """
    # Implementation details...
    return results
 
# Main support agent
@tracker.start_agent_tracing(key="support-agent")
async def handle_support_request(user_message: str, user_context: dict) -> str:
    """
    Handle a customer support request.
    
    Args:
        user_message (str): User's message
        user_context (dict): User's context and history
        
    Returns:
        str: Support agent's response
    """
    try:
        # Search knowledge base
        docs = await search_knowledge_base(user_message)
        
        # Generate response
        response = await generate_response(user_message, docs, user_context)
        
        return response
    except Exception as e:
        # Errors are automatically tracked
        raise
 
@tracker.trace_agent_node("response-generator")
async def generate_response(message: str, docs: list, context: dict) -> str:
    """
    Generate a support response based on knowledge base results.
    
    Args:
        message (str): User's message
        docs (list): Relevant documents
        context (dict): User context
        
    Returns:
        str: Generated response
    """
    # Implementation details...
    return response

3. Document Analysis Pipeline

document_analyzer.py


from handit_service import tracker
from langchain.chat_models import ChatOpenAI
from langchain.embeddings import OpenAIEmbeddings
import asyncio
 
# Configure the tracker
tracker.config(api_key="your-api-key")
 
# Initialize and track models
llm = tracker.track_model(
    ChatOpenAI(model="gpt-4"),
    model_id="document-gpt4"
)
embeddings = tracker.track_model(
    OpenAIEmbeddings(),
    model_id="document-embeddings"
)
 
# Main document analysis pipeline
@tracker.start_agent_tracing(key="document-analyzer")
async def analyze_document(document: dict) -> dict:
    """
    Analyze a document using multiple steps.
    
    Args:
        document (dict): Document data including content and metadata
        
    Returns:
        dict: Analysis results
    """
    try:
        # Process document in parallel
        tasks = [
            extract_entities(document),
            classify_document(document),
            summarize_content(document)
        ]
        
        # Wait for all tasks to complete
        results = await asyncio.gather(*tasks)
        
        # Combine results
        analysis = combine_results(results)
        
        return analysis
    except Exception as e:
        # Errors are automatically tracked
        raise
 
@tracker.trace_agent_node("entity-extractor")
async def extract_entities(document: dict) -> dict:
    """
    Extract named entities from document.
    
    Args:
        document (dict): Document data
        
    Returns:
        dict: Extracted entities
    """
    # Implementation details...
    return entities
 
@tracker.trace_agent_node("document-classifier")
async def classify_document(document: dict) -> dict:
    """
    Classify document type and category.
    
    Args:
        document (dict): Document data
        
    Returns:
        dict: Classification results
    """
    # Implementation details...
    return classification
 
@tracker.trace_agent_node("content-summarizer")
async def summarize_content(document: dict) -> dict:
    """
    Generate document summary.
    
    Args:
        document (dict): Document data
        
    Returns:
        dict: Summary information
    """
    # Implementation details...
    return summary
 
@tracker.trace_agent_node("result-combiner")
def combine_results(results: list) -> dict:
    """
    Combine results from all analysis steps.
    
    Args:
        results (list): Results from all analysis steps
        
    Returns:
        dict: Combined analysis results
    """
    # Implementation details...
    return combined_results

4. API Request Tracking

api_tracker.py


from handit_service import tracker
import requests
 
# Configure the tracker
tracker.config(api_key="your-api-key")
 
# Intercept and track API requests
requests.post = tracker.intercept_requests(requests.post)
requests.get = tracker.intercept_requests(requests.get)
 
# Example API client
class APIClient:
    def __init__(self, base_url: str):
        self.base_url = base_url
    
    @tracker.trace_agent_node("api-client")
    async def make_request(self, endpoint: str, method: str, data: dict = None) -> dict:
        """
        Make an API request with tracking.
        
        Args:
            endpoint (str): API endpoint
            method (str): HTTP method
            data (dict, optional): Request data
            
        Returns:
            dict: API response
        """
        url = f"{self.base_url}/{endpoint}"
        
        try:
            if method.upper() == "POST":
                response = requests.post(url, json=data)
            else:
                response = requests.get(url)
                
            response.raise_for_status()
            return response.json()
        except Exception as e:
            # Errors are automatically tracked
            raise

5. Performance Monitoring

performance_monitor.py


from handit_service import tracker
import time
import statistics
from typing import Dict, List, Any
 
class PerformanceMonitor:
    def __init__(self, monitor_id: str):
        self.monitor_id = monitor_id
        self.metrics: List[float] = []
    
    @tracker.trace_agent_node("performance-monitor")
    async def track_operation(
        self,
        operation: str,
        data: Dict[str, Any]
    ) -> Dict[str, Any]:
        """
        Track operation performance.
        
        Args:
            operation (str): Operation name
            data (dict): Operation data
            
        Returns:
            dict: Operation results
        """
        start_time = time.time()
        try:
            # Execute operation
            result = await self._execute_operation(operation, data)
            
            # Calculate metrics
            execution_time = time.time() - start_time
            self.metrics.append(execution_time)
            
            # Calculate statistics
            stats = {
                "current": execution_time,
                "average": statistics.mean(self.metrics),
                "median": statistics.median(self.metrics),
                "min": min(self.metrics),
                "max": max(self.metrics)
            }
            
            # Track performance metrics
            tracker.capture_model(
                model_id=f"{self.monitor_id}-metrics",
                request_body={
                    "operation": operation,
                    "data": data
                },
                response_body={
                    "execution_time": execution_time,
                    "statistics": stats,
                    "status": "success"
                }
            )
            
            return result
        except Exception as e:
            execution_time = time.time() - start_time
            self.metrics.append(execution_time)
            
            # Track error metrics
            tracker.capture_model(
                model_id=f"{self.monitor_id}-error",
                request_body={
                    "operation": operation,
                    "data": data
                },
                response_body={
                    "execution_time": execution_time,
                    "error": str(e),
                    "status": "error"
                }
            )
            raise
    
    async def _execute_operation(self, operation: str, data: dict) -> dict:
        """
        Execute the monitored operation.
        
        Args:
            operation (str): Operation name
            data (dict): Operation data
            
        Returns:
            dict: Operation results
        """
        # Implementation details...
        return result

These examples demonstrate real-world applications of the Handit.ai Python SDK, showing how to effectively track and monitor AI applications in production environments.

Remember to replace "your-api-key" with your actual Handit.ai API key in production code.