Python SDK - Pi Labs SDK

The Pi Labs Python SDK provides convenient, type-safe access to the Pi Client REST API from any Python 3.8+ application. This guide covers installation, basic usage, configuration, async programming, and advanced features. Documentation is also available at the GitHub repository. The SDK supports a wide range of Python environments and frameworks:

Python Versions

Python 3.8+ with full type support and async capabilities

Web Frameworks

Django, Flask, FastAPI, and other WSGI/ASGI frameworks

Async Support

Full asyncio support with httpx and optional aiohttp backends

Data Science

Jupyter Notebooks, pandas integration, and ML workflows

Quickstart

Get your API key

Retrieve your API key from your account page.

Install the SDK

Install the SDK using your preferred package manager:

pip install withpi

Initialize the client

Create a new client instance with your API key. If api_key is not specified, the SDK will look for the WITHPI_API_KEY environment variable.

import os
from withpi import PiClient

pi = PiClient(
  api_key=os.environ.get("WITHPI_API_KEY"),
)

Make your first scoring request

Issue a request to Pi Scorer:

scores = pi.scoring_system.score(
    llm_input="What are some good day trips from Milan by train?",
    llm_output="Milan is an excellent hub for day trips by train. You can take a short ride to the stunning shores of Lake Como to explore picturesque towns like Bellagio and Varenna. Alternatively, the historic hilltop city of Bergamo is another charming and easily accessible option. For a spectacular alpine adventure, consider the Bernina Express scenic train journey through the Swiss Alps. Cities like Turin and Verona are also just an hour or two away via high-speed train.",
    scoring_spec=[
        {"question": "Does the response maintain a professional tone?"},
        {"question": "Did the response fulfill the intent of the user's query?"},
        {"question": "Did the response only present data relevant to the user's query?"}
    ],
)

print('Total Score:', scores.total_score)
print('Question Scores:', scores.question_scores)

You should receive a response with the scores Pi Scorer assigned to the generation.

Async Usage

The SDK provides full async support with AsyncPiClient. Simply import AsyncPiClient instead of PiClient and use await with each API call. By default, the async client uses httpx for HTTP requests. However, for improved concurrency performance you may also use aiohttp as the HTTP backend.

httpx
aiohttp

import os
import asyncio
from withpi import AsyncPiClient

pi = AsyncPiClient(
  api_key=os.environ.get("WITHPI_API_KEY"),  # This is the default and can be omitted
)

async def main() -> None:
  scores = await pi.scoring_system.score(
    llm_input="What are some good day trips from Milan by train?",
    llm_output="Milan is an excellent hub for day trips by train. You can take a short ride to the stunning shores of Lake Como to explore picturesque towns like Bellagio and Varenna.",
    scoring_spec=[
      {"question": "Does the response maintain a professional tone?"},
      {"question": "Did the response fulfill the intent of the user's query?"},
    ],
  )
  print(scores.total_score)

asyncio.run(main())

Functionality between the synchronous and asynchronous clients is otherwise identical.

Type Safety

The SDK includes comprehensive type definitions using TypedDicts for requests and Pydantic models for responses:

from withpi import PiClient

pi = PiClient()

# TypedDict provides autocomplete for request parameters
scoring_params = {
  "llm_input": "What are some good day trips from Milan by train?",
  "llm_output": "Milan is an excellent hub for day trips by train. You can take a short ride to the stunning shores of Lake Como to explore picturesque towns like Bellagio and Varenna.",
  "scoring_spec": [
    {
      "label": "Professional Tone",
      "question": "Does the response maintain a professional tone?"
    },
    {
      "label": "Intent Fulfillment", 
      "question": "Did the response fulfill the intent of the user's query?"
    }
  ],
}

# Pydantic model provides type-safe response handling
response = pi.scoring_system.score(**scoring_params)

# Access response properties with full autocomplete
print(f"Total Score: {response.total_score}")
print(f"Question Scores: {response.question_scores}")

# Convert to different formats
response_dict = response.to_dict()      # Convert to dictionary
response_json = response.to_json()      # Serialize to JSON

Configuration Options

Client Initialization

Configure the client with various options for your specific use case:

from withpi import PiClient
import httpx

pi = PiClient(
  timeout=30.0,           # 30 seconds (default is 1 minute)
  max_retries=3,          # Maximum retry attempts (default is 2)
  base_url="https://api.withpi.ai",  # Custom base URL
)

Timeouts

Configure request timeouts globally or per request:

Global Configuration
Per-Request Configuration

from withpi import PiClient
import httpx

# Configure timeout for all requests
pi = PiClient(
  timeout=20.0,  # 20 seconds
)

# More granular control
pi = PiClient(
  timeout=httpx.Timeout(60.0, read=5.0, write=10.0, connect=2.0),
)

Retries

The SDK automatically retries certain errors with exponential backoff:

Connection errors
408 Request Timeout
409 Conflict
429 Rate Limit
500+ Internal Server errors

Global Retry Configuration
Per-Request Retry Configuration

# Configure retries for all requests
pi = PiClient(
  max_retries=0,  # Disable retries
)

Error Handling

In the event of errors, SDK raises subclasses of APIConnectionError:

import withpi
from withpi import PiClient

pi = PiClient()

try:
    result = pi.scoring_system.score(
        llm_input="What are some good day trips from Milan by train?",
        llm_output="Milan is an excellent hub for day trips by train.",
        scoring_spec=[
            {"question": "Does the response maintain a professional tone?"},
            {"question": "Did the response fulfill the intent of the user's query?"},
        ],
    )
except withpi.APIConnectionError as e:
    print("The server could not be reached")
    print(e.__cause__)  # an underlying Exception, likely raised within httpx.
except withpi.RateLimitError as e:
    print("A 429 status code was received; we should back off a bit.")
except withpi.APIStatusError as e:
    print("Another non-200-range status code was received")
    print(e.status_code)
    print(e.response)

Error Types

Status Code	Error Type	Description
400	`BadRequestError`	Invalid request parameters
401	`AuthenticationError`	Invalid or missing API key
403	`PermissionDeniedError`	Insufficient permissions
404	`NotFoundError`	Resource not found
422	`UnprocessableEntityError`	Request validation failed
429	`RateLimitError`	Rate limit exceeded
500+	`InternalServerError`	Server-side errors
N/A	`APIConnectionError`	Network connectivity issues

Requests that time out throw an APITimeoutError and are automatically retried according to your retry configuration.

Advanced Usage

Accessing Raw Response Data

Access the underlying response object for headers and other metadata:

Raw Response Only
Streaming Response

from withpi import PiClient

pi = PiClient()

response = pi.scoring_system.with_raw_response.score(
    llm_input="What are some good day trips from Milan by train?",
    llm_output="Milan is an excellent hub for day trips by train.",
    scoring_spec=[{"question": "Does the response maintain a professional tone?"}],
)

print(response.headers.get('X-My-Header'))
print(response.status_code)

# Parse the response data
scoring_system = response.parse()
print(scoring_system.total_score)

Custom and Undocumented Endpoints

You can make requests to any endpoint, including undocumented ones:

import httpx

# Custom endpoint
response = pi.post(
    "/some/path",
    cast_to=httpx.Response,
    body={"some_prop": "foo"},
)

print(response.headers.get("x-foo"))

# Undocumented parameters
pi.scoring_system.score(
    llm_input="What are some good day trips from Milan by train?",
    llm_output="Milan is an excellent hub for day trips by train.",
    scoring_spec=[{"question": "Does the response maintain a professional tone?"}],
    # Extra parameters
    extra_body={"undocumented_param": "value"},
    extra_query={"debug": "true"},
    extra_headers={"X-Custom-Header": "value"},
)

HTTP Client Customization

Customize the underlying HTTP client for proxies, transports, and advanced functionality:

import httpx
from withpi import PiClient, DefaultHttpxClient

# Global configuration
pi = PiClient(
    base_url="http://my.test.server.example.com:8083",
    http_client=DefaultHttpxClient(
        proxy="http://my.test.proxy.example.com",
        transport=httpx.HTTPTransport(local_address="0.0.0.0"),
    ),
)

# Per-request configuration
pi.with_options(
    http_client=DefaultHttpxClient(proxy="http://different.proxy.com")
).scoring_system.score(
    llm_input="What are some good day trips from Milan by train?",
    llm_output="Milan is an excellent hub for day trips by train.",
    scoring_spec=[{"question": "Does the response maintain a professional tone?"}]
)

Resource Management

When making a one-off request, you can use a context manager to ensure the underlying HTTP client and its open connections are closed properly:

from withpi import PiClient

# Context manager (recommended)
with PiClient() as pi:
  result = pi.scoring_system.score(
    llm_input="What are some good day trips from Milan by train?",
    llm_output="Milan is an excellent hub for day trips by train.",
    scoring_spec=[{"question": "Does the response maintain a professional tone?"}]
  )
# HTTP client is now closed.

# To close manually:
pi = PiClient()
try:
  result = pi.scoring_system.score(
    # ...
  )
finally:
  pi.close()

Logging and Debugging

Logging can be enabled by setting the PI_CLIENT_LOG environment variable:

# Default logging
export PI_CLIENT_LOG=info

# More verbose logging
export PI_CLIENT_LOG=debug

You can also check the installed version with __version__:

import withpi
print(withpi.__version__)

Handling Null vs Missing Fields

In an API response, a field may be explicitly null, or missing entirely; in either case, its value is None in this library. You can differentiate the two cases with .model_fields_set:

if response.my_field is None:
    if 'my_field' not in response.model_fields_set:
        print('Got json like {}, without a "my_field" key present at all.')
    else:
        print('Got json like {"my_field": null}.')

Next Steps

API Reference

Explore all available endpoints and their parameters

Writing Rubrics

Learn how to evaluate quality with Pi Scorer

JavaScript SDK

Call Pi endpoints from JavaScript

Integrations

Connect your low-code or no-code tools to Pi

Getting Started

SDK

API Reference

Python Versions

Web Frameworks

Async Support

Data Science

​Quickstart

​Async Usage

​Type Safety

​Configuration Options

​Client Initialization

​Timeouts

​Retries

​Error Handling

​Error Types

​Advanced Usage

​Accessing Raw Response Data

​Custom and Undocumented Endpoints

​HTTP Client Customization

​Resource Management

​Logging and Debugging

​Handling Null vs Missing Fields

​Next Steps

API Reference

Writing Rubrics

JavaScript SDK

Integrations

Quickstart

Async Usage

Type Safety

Configuration Options

Client Initialization

Timeouts

Retries

Error Handling

Error Types

Advanced Usage

Accessing Raw Response Data

Custom and Undocumented Endpoints

HTTP Client Customization

Resource Management

Logging and Debugging

Handling Null vs Missing Fields

Next Steps