Python SDK Reference

Complete API reference for the TuringPulse Python SDK.

Python 3.8+Async Support

Installation

Installation
pip install turingpulse_sdk

# Or with specific integrations
pip install turingpulse_sdk[langgraph]
pip install turingpulse_sdk[langchain]
pip install turingpulse_sdk[all]

Initialization

init()

Initialize the TuringPulse SDK. Call this once at application startup.

init.py
from turingpulse_sdk import init, TuringPulseConfig, FingerprintConfig

init(TuringPulseConfig(
    api_key="sk_live_...",                  # Required (or TP_API_KEY env var)
    workflow_name="my-project",             # Required (or TP_WORKFLOW_NAME env var)
    
    # Endpoint (optional — defaults to https://api.turingpulse.ai)
    # endpoint="https://api.turingpulse.ai",  # Or TP_ENDPOINT env var
    
    # Data Capture
    capture_arguments=False,                # Capture function arguments
    capture_return_value=False,             # Capture return values
    
    # Security — redact sensitive fields before telemetry leaves your environment
    redact_fields=["password", "api_key", "secret", "token"],
    
    # Fingerprinting — detect prompt/config drift automatically
    fingerprint=FingerprintConfig(
        enabled=True,
        capture_prompts=True,
        capture_configs=True,
    ),
))

Parameters

ParameterTypeDefaultDescription
api_keystrTP_API_KEY envYour API key (required).
workflow_namestrTP_WORKFLOW_NAME envWorkflow name for tracing (required).
endpointstrhttps://api.turingpulse.aiAPI endpoint (or TP_ENDPOINT env var).
capture_argumentsboolFalseCapture function arguments in traces.
capture_return_valueboolFalseCapture return values in traces.
redact_fieldslist[str][]Field names to redact from telemetry.

Instrumentation

@instrument()

Decorator to trace and instrument a function. Creates a trace with spans automatically.

from turingpulse_sdk import instrument

@instrument(
    name="my-agent",              # Unique agent/workflow identifier
    operation="handle_query",         # Optional operation name
    labels={"team": "support"},       # Custom labels for filtering
    trace=True,                       # Enable tracing (default: True)
    metadata={"version": "1.0"},      # Static metadata attached to every run
)
def my_agent(query: str) -> str:
    return process(query)

# Async functions are supported automatically
@instrument(name="async-agent")
async def my_async_agent(query: str) -> str:
    return await process_async(query)

current_context()

Access the active execution context from anywhere in your code.

from turingpulse_sdk import current_context

def nested_function():
    ctx = current_context()
    if ctx:
        # Access run_id, parent span, depth, etc.
        print(f"Current run: {ctx.run_id}")

Evaluations

Evaluations (heuristic, LLM-as-Judge, and custom metrics) are configured and triggered through the TuringPulse platform, not the SDK. Navigate to Analysis → Evaluations in the UI or use the CLI:

# Run evaluations via CLI
tp evals run --config eval-config-id --workflow my-agent

# Or via REST API
# POST https://api.turingpulse.ai/v1/evals/evaluate

Feedback

User feedback is submitted via the REST API, not the SDK directly:

# POST https://api.turingpulse.ai/v1/feedback
# {
#   "trace_id": "trace-123",
#   "score": 0.8,
#   "feedback_type": "rating",
#   "comment": "Very helpful response"
# }

Governance

@instrument() with Governance

from turingpulse_sdk import instrument, GovernanceDirective

@instrument(
    name="high-stakes-agent",
    governance=GovernanceDirective(
        hitl=True,
        reviewers=["admin@company.com"],
        escalation_channels=["pagerduty://critical"],
        auto_escalate_after_seconds=3600,
        severity="high",
    )
)
def execute_trade(order: dict):
    return trading_api.execute(order)

Deploy Tracking

register_deploy()

Register a deployment for change correlation.

from turingpulse_sdk import register_deploy

# Auto-detect from CI/CD environment
register_deploy(
    workflow_id="my-agent",
    auto_detect=True,
)

# Or provide explicit values
register_deploy(
    workflow_id="my-agent",
    version="1.2.3",
    git_sha="abc123",
    commit_message="Fix prompt template",
)

Utilities

shutdown()

Gracefully shutdown the SDK plugin. Call at application exit.

from turingpulse_sdk import get_plugin

plugin = get_plugin()
plugin.shutdown()

Next Steps