Overview

Pioneer pip package

The Personalization Client provides a lightweight Python interface for interacting with the Fastino Personalization API. It handles authentication, request validation, error handling, and response parsing — allowing developers to integrate the Personalization API with minimal boilerplate.

Before You Start

Requirements

Before installing the Personalization Client, ensure you have:

Installing the Client

You can install the client in three ways.

Official Package Index (Recommended)

pip install --index-url https://fastino-ai.github.io/Pioneer/simple/ personalization-client

Setting Up Authentication

The client automatically loads your API key from the environment using:

PIONEER_API_KEY=your-api-key

Option A: Environment Variable (Recommended)

# macOS / Linux
export PIONEER_API_KEY='your-api-key-here'

# Windows (PowerShell)
$env:PIONEER_API_KEY='your-api-key-here'

# Windows (Command Prompt)
set PIONEER_API_KEY=your-api-key-here

Option B: Using a .env File

.env:

PIONEER_API_KEY=your-api-key-here

Python:

from dotenv import load_dotenv
from personalization_client import PersonalizationClient

load_dotenv()
client = PersonalizationClient()

Option C: Direct Key (Not Recommended)

from personalization_client import PersonalizationClient

client = PersonalizationClient(api_key="your-api-key-here")

Quick Start

Below is a complete example demonstrating the core functionality of the client.

from personalization_client import PersonalizationClient

# Initialize client
client = PersonalizationClient()

# Register a user
response = client.register_user(
    email="user@example.com",
    traits={"name": "John Doe"}
)
user_id = response.user_id

# Ingest conversation data
client.ingest_data(
    user_id=user_id,
    source="slack",
    message_history=[
        {"role": "user", "content": "I love hiking", "timestamp": "2025-01-01T10:00:00Z"}
    ]
)

# Fetch the user's profile summary
summary = client.get_summary(user_id=user_id)
print(summary.summary)

# Retrieve relevant chunks
chunks = client.get_chunks(
    user_id=user_id,
    conversation_history=[{"role": "user", "content": "What are my interests?"}]
)

# Delete the user
client.delete_user(user_id=user_id)

Client API Reference

1. Register a User

Initializes a personalization profile.

Method: register_user()

register_user(
  email: str,
  purpose: Optional[str] = None,
  traits: Optional[Dict[str, Any]] = None,
  callback_url: Optional[str] = None
) -> RegisterUserResponse

Parameters

Field

Type

Description

email

str

Required. User’s email address

purpose

str

Optional purpose (e.g., "customer_support", "analytics")

traits

dict

Optional traits (name, timezone, social URLs)

callback_url

str

Optional webhook fired after completion

Example

client = PersonalizationClient()

response = client.register_user(
    email="jane.smith@example.com",
    purpose="customer_support",
    traits={
        "name": "Jane Smith",
        "linkedin_url": "https://linkedin.com/in/janesmith",
        "timezone": "America/New_York"
    },
    callback_url="https://yourapp.com/webhooks/registration"
)

print(response.user_id)
print(response.status)
print(response.stages_completed)

2. Ingest User Data

Adds conversations and/or documents to the personalization engine. Data is anonymized via GLiNER-2.

Method: ingest_data()

ingest_data(
  user_id: str,
  source: Optional[str] = None,
  message_history: Optional[list] = None,
  documents: Optional[list] = None,
  dedupe: bool = False
) -> UnifiedIngestResponse

Example: Messages

response = client.ingest_data(
    user_id="user-123",
    source="email",
    message_history=[
        {"role": "user", "content": "I have a meeting tomorrow", "timestamp": "2025-01-15T10:00:00Z"}
    ]
)

Example: Documents

response = client.ingest_data(
    user_id="user-123",
    source="notes",
    documents=[
        {
            "content": "Project requirements document...",
            "title": "Project Requirements",
            "document_type": "note",
            "created_at": "2025-01-15T09:00:00Z"
        }
    ],
    dedupe=True
)

3. Get Profile Summary

Fetch a natural-language summary of the user — ideal for system prompts.

Method: get_summary()

get_summary(
  user_id: Optional[str] = None,
  user_email: Optional[str] = None,
  purpose: Optional[str] = None,
  max_chars: int = 1000,
  freshness_days: int = 365
) -> ProfileSummaryResponse

Example

summary = client.get_summary(user_id="user-123", max_chars=500)
print(summary.summary)

4. Retrieve Relevant Chunks

Unified vector retrieval and memory.

Method: get_chunks()

get_chunks(
  user_id: str,
  conversation_history: list,
  k: int = 6,
  max_context_turns: int = 4,
  exclude_chunk_ids: Optional[list] = None,
  similarity_threshold: float = 0.25
) -> ProfileChunksResponse

Example

chunks = client.get_chunks(
    user_id="user-123",
    conversation_history=[
        {"role": "user", "content": "Tell me about my work habits"}
    ],
    k=10,
    similarity_threshold=0.3
)

5. Delete a User

Deletes a user and all associated data.

Method: delete_user()

delete_user(user_id: str) -> DeleteUserResponse

Example

response = client.delete_user("user-123")
print(response.status)
print(response.message)

Example Response

{
  "success": true,
  "message": "User <uuid> deleted successfully",
  "user_id": "uuid-of-user-deleted"
}

Bonus: Health Check

Verify API connectivity.

health = client.health()
print(health)

Error Handling

The client exposes structured exception classes consistent with API behavior.

Exception Hierarchy

  • PersonalizationAPIError

    • AuthenticationError (401)

    • ValidationError (400 / 422)

    • NotFoundError (404)

    • ServerError (500)

Example

try:
    client.register_user(email="invalid-email")
except ValidationError as e:
    print("Invalid request:", e.message)

Context Manager Support

The client can automatically clean up connections:

with PersonalizationClient() as client:
    user = client.register_user(email="user@example.com")

On this page