toothfairyai 0.8.1

Official Python SDK for ToothFairyAI API

ToothFairyAI Python SDK

Official Python SDK for the ToothFairyAI API - A comprehensive toolkit for building AI-powered applications with chat, document processing, agent management, and more.

Status: ✅ Production Ready | Version: 0.5.9

========================= test session starts ==========================
platform darwin -- Python 3.13.1, pytest-9.0.1

tests/debug/test_agent_functions.py     ✓✓✓✓     4 passed
tests/debug/test_agent_update.py        ✓✓✓✓     4 passed
tests/debug/test_authorisations.py      ✓✓✓✓✓✓✓  7 passed
tests/debug/test_members.py             ✓✓       2 passed
tests/debug/test_secrets.py             ✓        1 passed

========================= 18 passed in 35.2s ===========================

Installation

pip install toothfairyai

Quick Start

from toothfairyai import ToothFairyClient

# Initialize the client
client = ToothFairyClient(
    api_key="your-api-key",
    workspace_id="your-workspace-id"
)

# Send a message to an agent (non-streaming)
response = client.chat.send_to_agent(
    message="Hello, how can you help me?",
    agent_id="your-agent-id"
)
print(response.agent_response)

Table of Contents

• SDK Overview
• Client Configuration
• Core Features
• Validation Models
• Complete API Reference
  • Chat & Messaging
  • Agents
  • Documents
  • Entities
  • Folders
  • Prompts
  • Agent Functions
  • Authorisations
  • Channels
  • Connections
  • Members
  • Sites
  • Benchmarks
  • Hooks
  • Scheduled Jobs
  • Secrets
  • Dictionary
  • Embeddings
  • Settings
  • Billing
  • Streams
  • Request Logs
• Streaming Responses
• Error Handling
• Best Practices

SDK Overview

The ToothFairyAI SDK provides a Pythonic interface to all 95 API endpoints, organized into 18 specialized managers:

Manager	Purpose	Endpoints
client.chat	Chat conversations & messages	7
client.agents	AI agent management	5
client.documents	Document upload & management	6
client.entities	Topics, intents, NER entities	5
client.folders	Content organization	5
client.prompts	Prompt templates	5
client.agent_functions	Agent function calls	5
client.authorisations	Auth configurations	5
client.channels	Communication channels	5
client.connections	Database/API connections	3
client.members	Workspace members	4
client.sites	Website crawling	4
client.benchmarks	Performance testing	5
client.hooks	Custom code execution	5
client.scheduled_jobs	Cron jobs & scheduling	5
client.secrets	Secret management	2
client.dictionary	Translation dictionary	2
client.embeddings	Document embeddings	1
client.charting_settings	Charting configuration	2
client.embeddings_settings	Embeddings configuration	2
client.billing	Usage & costs	1
client.streams	Output streams	2
client.request_logs	Request logging	2
client.streaming	Real-time streaming	-

Client Configuration

from toothfairyai import ToothFairyClient

client = ToothFairyClient(
    api_key="your-api-key",           # Required: Your API key
    workspace_id="your-workspace-id", # Required: Your workspace ID
    base_url="https://api.toothfairyai.com",  # Optional: API base URL
    ai_url="https://ai.toothfairyai.com",     # Optional: AI endpoint URL
    ai_stream_url="https://ais.toothfairyai.com",  # Optional: Streaming URL
    timeout=120  # Optional: Request timeout in seconds (default: 120)
)

Getting Your Credentials

1. API Key: Log in to ToothFairyAI Admin → API Integration → Generate API Key
2. Workspace ID: Found in your workspace settings or API dashboard

Core Features

1. Chat & Messaging

Create conversations, send messages, and manage chat history with AI agents.

2. Agent Management

Create, configure, and manage AI agents with different modes (retriever, chatter, planner, etc.).

3. Document Processing

Upload, search, and manage documents for knowledge base integration.

4. Entity Management

Organize content with topics, intents, and named entity recognition.

5. Streaming

Real-time streaming responses with an iterator pattern similar to OpenAI's SDK.

6. Validation Models (NEW)

Pre-validate API payloads before sending requests using Pydantic models. Catch errors early and get clear error messages.

Complete API Reference

Chat & Messaging

Create a Chat Session

# Create a new chat
chat = client.chat.create(
    name="Customer Support",
    customer_id="customer-123",
    customer_info={"name": "John Doe", "email": "[e-mail]"},
    primary_role="user",
    external_participant_id="[e-mail]",
    channel_settings={
        "sms": {"isEnabled": True, "recipient": "[phone]"},
        "email": {"isEnabled": True, "recipient": "[e-mail]"}
    }
)
print(f"Chat ID: {chat.id}")

Update a Chat

chat = client.chat.update(
    chat_id="chat-id",
    name="Updated Chat Name",
    customer_id="new-customer-id"
)

Get Chat Details

chat = client.chat.get(chat_id="chat-id")
print(f"Chat: {chat.name}")
print(f"Primary Role: {chat.primary_role}")

List Chats

chats = client.chat.list(limit=10, offset=0)
for chat in chats.items:
    print(f"{chat.name} (ID: {chat.id})")

Delete a Chat

client.chat.delete(chat_id="chat-id")

Create a Message

message = client.chat.create_message(
    chat_id="chat-id",
    text="Hello, I need help with my account",
    role="user",
    user_id="user-123",
    images=["https://example.com/image.jpg"],
    files=["https://example.com/document.pdf"]
)

Update a Message

message = client.chat.update_message(
    message_id="message-id",
    text="Updated message text",
    role="user"
)

Get a Message

message = client.chat.get_message(message_id="message-id")
print(f"Message: {message.text}")

List Messages in a Chat

messages = client.chat.list_messages(
    chat_id="chat-id",
    limit=20,
    offset=0
)
for msg in messages.items:
    print(f"{msg.role}: {msg.text}")

Send Message to Agent (Non-Streaming)

response = client.chat.send_to_agent(
    message="Hello, how can you help me?",
    agent_id="agent-id",
    chat_id="existing-chat-id",  # Optional - creates new chat if not provided
    phone_number="[phone]",  # Optional
    customer_id="customer-123",   # Optional
    provider_id="provider-id",    # Optional
    customer_info={"name": "John"},  # Optional
    attachments={
        "images": ["https://example.com/image.jpg"],
        "files": ["https://example.com/document.pdf"],
        "audios": ["https://example.com/audio.mp3"],
        "videos": ["https://example.com/video.mp4"]
    }
)
print(f"Response: {response.agent_response}")
print(f"Chat ID: {response.chat_id}")
print(f"Message ID: {response.message_id}")

Agents

Create an Agent

agent = client.agents.create(
    label="Customer Support Agent",
    mode="retriever",  # retriever, coder, chatter, planner, computer, voice
    interpolation_string="You are a helpful customer support assistant...",
    goals="Help customers resolve issues and answer questions",
    temperature=0.7,
    max_tokens=2000,
    max_history=10,
    top_k=10,
    doc_top_k=5,
    description="Specialized agent for customer support",
    has_memory=True,
    show_citations=True,
    allowed_topics=["topic-1", "topic-2"],
    static_docs=["doc-1", "doc-2"]
)
print(f"Agent ID: {agent.id}")

Create an Agent with Structured JSON Output

# Define JSON schema for structured responses
json_schema = {
    "type": "object",
    "properties": {
        "summary": {"type": "string"},
        "key_points": {"type": "array", "items": {"type": "string"}},
        "sentiment": {"type": "string", "enum": ["positive", "neutral", "negative"]}
    },
    "required": ["summary", "key_points", "sentiment"]
}

agent = client.agents.create(
    label="Analysis Agent",
    mode="retriever",  # works with retriever or chatter mode
    interpolation_string="You are an assistant that returns structured JSON output.",
    goals="Analyze text and return structured analysis.",
    temperature=0.3,
    max_tokens=1000,
    max_history=5,
    top_k=5,
    doc_top_k=3,
    enable_json_output=True,
    json_output_structure=json_schema
)
# Agent will always respond with JSON matching the schema

Update an Agent

agent = client.agents.update(
    agent_id="agent-id",
    label="Updated Agent Name",
    temperature=0.8,
    max_tokens=3000,
    description="Updated description",
    charting=False  # Required field, defaults to False
)

Get Agent Details

agent = client.agents.get(agent_id="agent-id")
print(f"Agent: {agent.label}")
print(f"Mode: {agent.mode}")
print(f"Goals: {agent.goals}")

List Agents

agents = client.agents.list(limit=10, offset=0)
for agent in agents.items:
    print(f"{agent.label} ({agent.mode})")

Delete an Agent

client.agents.delete(agent_id="agent-id")

Documents

Upload a Document

# Upload with progress callback
result = client.documents.upload(
    file_path="./document.pdf",
    folder_id="folder-id",
    on_progress=lambda percent, loaded, total: print(f"Progress: {percent}%")
)
print(f"Uploaded: {result.filename}")
print(f"Size: {result.size_in_mb:.2f} MB")

Upload from Base64

import base64

with open("document.pdf", "rb") as f:
    base64_data = base64.b64encode(f.read()).decode()

result = client.documents.upload_from_base64(
    base64_data=base64_data,
    filename="document.pdf",
    content_type="application/pdf",
    folder_id="folder-id"
)

Create Document from URL

doc = client.documents.create_from_path(
    file_path="https://example.com/document.pdf",
    user_id="user-123",
    title="External Document",
    folder_id="folder-id",
    topics=["topic-1", "topic-2"],
    status="published"
)

Create Document Record

doc = client.documents.create(
    user_id="user-123",
    title="My Document",
    doc_type="readComprehensionFile",  # readComprehensionUrl, readComprehensionPdf
    topics=["topic-1"],
    folder_id="folder-id",
    external_path="https://example.com/doc.pdf",
    source="external",
    status="published",
    scope="training"
)

Get Document

doc = client.documents.get(document_id="doc-id")
print(f"Title: {doc.title}")
print(f"Type: {doc.doc_type}")
print(f"Status: {doc.status}")

Update Document

doc = client.documents.update(
    document_id="doc-id",
    user_id="user-123",
    title="Updated Title",
    topics=["new-topic"],
    folder_id="new-folder-id",
    status="published"
)

List Documents

docs = client.documents.list(
    limit=20,
    offset=0,
    folder_id="folder-id",  # Optional filter
    status="published"      # Optional filter
)
for doc in docs.items:
    print(f"{doc.title} (ID: {doc.id})")

Delete Document

client.documents.delete(document_id="doc-id")

Search Documents

results = client.documents.search(
    text="product features",
    top_k=10,
    metadata={"category": "procedures"}
)
for result in results:
    print(f"Score: {result.get('score')}")
    print(f"Content: {result.get('content')}")

Download Document

result = client.documents.download(
    filename="document.pdf",
    output_path="./downloads/document.pdf",
    context="documents",
    on_progress=lambda p, l, t: print(f"Download: {p}%")
)
print(f"Downloaded: {result.filename}")

Entities

Create an Entity

entity = client.entities.create(
    user_id="user-123",
    label="Product Support",
    entity_type="topic",  # topic, intent, ner
    description="Customer product support inquiries",
    emoji="💡",
    parent_entity="parent-topic-id",
    background_color="#FF5733"
)

Get Entity

entity = client.entities.get(entity_id="entity-id")
print(f"Label: {entity.label}")
print(f"Type: {entity.entity_type}")

Update Entity

entity = client.entities.update(
    entity_id="entity-id",
    label="Updated Label",
    description="New description",
    emoji="🔧"
)

Delete Entity

client.entities.delete(entity_id="entity-id")

List Entities

entities = client.entities.list(limit=20, entity_type="topic")
for entity in entities.items:
    print(f"{entity.label} ({entity.entity_type})")

Get Entities by Type

topics = client.entities.get_by_type("topic")
intents = client.entities.get_by_type("intent")
ner_entities = client.entities.get_by_type("ner")

Search Entities

results = client.entities.search("cleaning", entity_type="topic")
for entity in results:
    print(f"{entity.label}")

Folders

Create a Folder

folder = client.folders.create(
    user_id="user-123",
    name="Procedures",
    description="Medical procedures documentation",
    emoji="📁",
    status="active",
    parent="parent-folder-id"  # Optional - for subfolders
)

Get Folder

folder = client.folders.get(folder_id="folder-id")
print(f"Name: {folder.name}")
print(f"Parent: {folder.parent}")

Update Folder

folder = client.folders.update(
    folder_id="folder-id",
    name="Updated Name",
    description="New description",
    status="active"
)

Delete Folder

client.folders.delete(folder_id="folder-id")

List Folders

folders = client.folders.list(
    status="active",
    limit=20,
    offset=0
)
for folder in folders.items:
    print(f"{folder.name}")

Get Root Folders

root_folders = client.folders.get_root_folders()
for folder in root_folders:
    print(f"Root: {folder.name}")

Get Subfolders

subfolders = client.folders.get_subfolders(parent_id="folder-id")
for folder in subfolders:
    print(f"Subfolder: {folder.name}")

Get Folder Tree

tree = client.folders.get_tree()
def print_tree(nodes, level=0):
    for node in nodes:
        print("  " * level + f"📁 {node.name}")
        print_tree(node.children, level + 1)

print_tree(tree)

Search Folders

results = client.folders.search("procedures")
for folder in results:
    print(f"{folder.name}")

Prompts

Create a Prompt

prompt = client.prompts.create(
    user_id="user-123",
    label="Greeting",
    prompt_type="greeting",
    interpolation_string="Hello {{name}}, how can I help you today?",
    scope="global",
    style="friendly",
    domain="customer-support",
    prompt_placeholder="Enter greeting message",
    available_to_agents=["agent-1", "agent-2"]
)

Get Prompt

prompt = client.prompts.get(prompt_id="prompt-id")
print(f"Label: {prompt.label}")
print(f"Template: {prompt.interpolation_string}")

Update Prompt

prompt = client.prompts.update(
    prompt_id="prompt-id",
    label="Updated Label",
    interpolation_string="New template string",
    available_to_agents=["agent-3"]
)

Delete Prompt

client.prompts.delete(prompt_id="prompt-id")

List Prompts

prompts = client.prompts.list(
    prompt_type="greeting",
    limit=20,
    offset=0
)
for prompt in prompts.items:
    print(f"{prompt.label}")

Get Prompts by Type

greetings = client.prompts.get_by_type("greeting")

Get Prompts by Agent

agent_prompts = client.prompts.get_by_agent("agent-id")

Get Prompts by Scope

global_prompts = client.prompts.get_by_scope("global")

Search Prompts

results = client.prompts.search("greeting", prompt_type="greeting")

Clone Prompt

cloned = client.prompts.clone(
    prompt_id="prompt-id",
    user_id="user-123",
    label="Cloned Greeting",
    interpolation_string="Modified template"
)

Agent Functions

Create an Agent Function

# Create a simple GET function
function = client.agent_functions.create(
    name="Weather API",
    description="Get current weather data for a location",  # Required
    url="https://api.weather.com/current",
    request_type="GET",
    authorisation_type="bearer",
    authorisation_key="your-token",
    parameters=[
        {"name": "location", "type": "string", "required": True}
    ]
)
print(f"Function ID: {function.id}")

# Create a POST function with headers
function = client.agent_functions.create(
    name="Create Order",
    description="Create a new order in the system",  # Required
    url="https://api.example.com/orders",
    request_type="POST",
    authorisation_type="apikey",
    headers=[{"key": "Content-Type", "value": "application/json"}],
    static_args=[{"key": "source", "value": "chatbot"}]
)

Update Agent Function

function = client.agent_functions.update(
    agent_function_id="function-id",
    name="Updated Function Name",
    url="https://new-url.com"
)

Delete Agent Function

client.agent_functions.delete(agent_function_id="function-id")

Get Agent Function

function = client.agent_functions.get(agent_function_id="function-id")
print(f"Name: {function.name}")
print(f"URL: {function.url}")

List Agent Functions

functions = client.agent_functions.list(limit=20)
for func in functions.items:
    print(f"{func.name}")

# Search by name
results = client.agent_functions.search("weather")

Authorisations

Create an Authorisation

# API Key authorisation
auth = client.authorisations.create(
    name="External API Key",
    auth_type="apikey",  # apikey, bearer, oauth, basic, none, username_and_password, env
    description="API key for external service",
    token_secret="your-api-key-value"
)
print(f"Auth ID: {auth.id}")

# Bearer token authorisation
auth = client.authorisations.create(
    name="Service Token",
    auth_type="bearer",
    token_secret="your-bearer-token"
)

# OAuth authorisation
auth = client.authorisations.create(
    name="GitHub OAuth",
    auth_type="oauth",
    description="GitHub OAuth for API access",
    scope="repo,user",
    grant_type="authorization_code",
    client_id="github-client-id",
    client_secret="encrypted-secret",
    authorization_base_url="https://github.com/login/oauth/authorize",
    token_base_url="https://github.com/login/oauth/access_token"
)

# Basic auth
auth = client.authorisations.create(
    name="Basic Auth",
    auth_type="basic",
    description="Basic authentication credentials"
)

Update Authorisation

# Note: auth_type cannot be changed after creation
auth = client.authorisations.update(
    authorisation_id="auth-id",
    name="Updated Auth Name",
    description="New description"
)

Delete Authorisation

client.authorisations.delete(authorisation_id="auth-id")

Get Authorisation

auth = client.authorisations.get(authorisation_id="auth-id")
print(f"Name: {auth.name}")
print(f"Type: {auth.type}")

List Authorisations

auths = client.authorisations.list(limit=20)
for auth in auths.items:
    print(f"{auth.name} ({auth.type})")

# Get authorisations by type
api_keys = client.authorisations.get_by_type("apikey")
oauth_auths = client.authorisations.get_by_type("oauth")

# Search by name
results = client.authorisations.search("github")

Channels

Create a Channel

channel = client.channels.create(
    name="SMS Channel",
    channel="sms",
    provider="twilio",
    senderid="[phone]",
    description="SMS notifications via Twilio",
    is_active=True
)
print(f"Channel ID: {channel.id}")

Update Channel

channel = client.channels.update(
    channel_id="channel-id",
    name="Updated Channel",
    is_active=False
)

Delete Channel

client.channels.delete(channel_id="channel-id")

Get Channel

channel = client.channels.get(channel_id="channel-id")
print(f"Name: {channel.name}")
print(f"Provider: {channel.provider}")

List Channels

channels = client.channels.list(limit=20)
for channel in channels.items:
    print(f"{channel.name} ({channel.provider})")

Connections

Get Connection

connection = client.connections.get(connection_id="connection-id")
print(f"Name: {connection.name}")
print(f"Type: {connection.type}")
print(f"Host: {connection.host}")

List Connections

connections = client.connections.list(limit=20)
for conn in connections.items:
    print(f"{conn.name} ({conn.type})")

Delete Connection

client.connections.delete(connection_id="connection-id")

Members

Get Member

member = client.members.get(member_id="member-id")
print(f"User ID: {member.user_id}")
print(f"Role: {member.role}")
print(f"Status: {member.status}")

List Members

members = client.members.list(limit=20)
for member in members.items:
    print(f"{member.user_id} ({member.role})")

Update Member

member = client.members.update(
    member_id="member-id",
    role="admin",
    status="active"
)

Delete Member

client.members.delete(member_id="member-id")

Sites

Get Site

site = client.sites.get(site_id="site-id")
print(f"Name: {site.name}")
print(f"URL: {site.url}")
print(f"Status: {site.status}")

List Sites

sites = client.sites.list(limit=20)
for site in sites.items:
    print(f"{site.name} - {site.url}")

Update Site

site = client.sites.update(
    site_id="site-id",
    name="Updated Site",
    status="active",
    allowed_paths=["/docs", "/blog"]
)

Delete Site

client.sites.delete(site_id="site-id")

Benchmarks

Create a Benchmark

benchmark = client.benchmarks.create(
    name="Customer Support Test",
    description="Test agent performance on customer support queries",
    questions=[
        {
            "question": "How do I reset my password?",
            "answer": "Go to Settings > Security > Reset Password",
            "reasoning": "Standard password reset flow",
            "context": "Account management"
        }
    ],
    files=["doc-1", "doc-2"]
)
print(f"Benchmark ID: {benchmark.id}")

Update Benchmark

benchmark = client.benchmarks.update(
    benchmark_id="benchmark-id",
    name="Updated Benchmark",
    description="New description"
)

Delete Benchmark

client.benchmarks.delete(benchmark_id="benchmark-id")

Get Benchmark

benchmark = client.benchmarks.get(benchmark_id="benchmark-id")
print(f"Name: {benchmark.name}")
print(f"Questions: {len(benchmark.questions)}")

List Benchmarks

benchmarks = client.benchmarks.list(limit=20)
for benchmark in benchmarks.items:
    print(f"{benchmark.name}")

Hooks

Create a Hook

hook = client.hooks.create(
    name="Data Processing Hook",
    function_name="process_data",
    custom_execution_code="def process_data(data): return data.upper()",
    custom_execution_instructions="Process incoming data",
    available_libraries="pandas,numpy",
    allow_external_api=True,
    hardcoded_script=False
)
print(f"Hook ID: {hook.id}")

Update Hook

hook = client.hooks.update(
    hook_id="hook-id",
    name="Updated Hook",
    custom_execution_code="def process_data(data): return data.lower()"
)

Delete Hook

client.hooks.delete(hook_id="hook-id")

Get Hook

hook = client.hooks.get(hook_id="hook-id")
print(f"Name: {hook.name}")
print(f"Function: {hook.function_name}")

List Hooks

hooks = client.hooks.list(limit=20)
for hook in hooks.items:
    print(f"{hook.name}")

Scheduled Jobs

Create a Scheduled Job

job = client.scheduled_jobs.create(
    name="Daily Report",
    description="Generate daily report at 9 AM",
    agent_id="agent-id",
    custom_prompt_id="prompt-id",
    forced_prompt="Generate a daily summary report",
    schedule={
        "frequency": "DAILY",
        "hour": 9,
        "minute": 0
    },
    timezone="UTC",
    is_active=True,
    status="ACTIVE"
)
print(f"Job ID: {job.id}")

Update Scheduled Job

job = client.scheduled_jobs.update(
    scheduled_job_id="job-id",
    name="Updated Job",
    is_active=False,
    schedule={"frequency": "WEEKLY", "dayOfWeek": 1, "hour": 9}
)

Delete Scheduled Job

client.scheduled_jobs.delete(scheduled_job_id="job-id")

Get Scheduled Job

job = client.scheduled_jobs.get(scheduled_job_id="job-id")
print(f"Name: {job.name}")
print(f"Status: {job.status}")
print(f"Schedule: {job.schedule}")

List Scheduled Jobs

jobs = client.scheduled_jobs.list(limit=20)
for job in jobs.items:
    print(f"{job.name} ({job.status})")

Secrets

Secrets are linked to authorisations. You must first create an authorisation, then create a secret for it.

Create a Secret

# First, create an authorisation
auth = client.authorisations.create(
    name="External Service Auth",
    auth_type="apikey",
    description="Authorisation for external service"
)

# Then create a secret linked to that authorisation
secret = client.secrets.create(
    authorisation_id=auth.id,
    password_secret_value="your-secret-api-key-12345"
)
print(f"Secret created: {secret.name}")

Delete Secret

client.secrets.delete(secret_id="secret-id")

Dictionary

Get Dictionary Entry

entry = client.dictionary.get(entry_id="entry-id")
print(f"Source: {entry.source_text}")
print(f"Target: {entry.target_text}")
print(f"Languages: {entry.source_language} -> {entry.target_language}")

List Dictionary Entries

entries = client.dictionary.list(limit=20)
for entry in entries.items:
    print(f"{entry.source_language} -> {entry.target_language}")

Embeddings

Get Document Embeddings

embeddings = client.embeddings.get()
for emb in embeddings:
    print(f"Document: {emb.title}")
    print(f"Chunk: {emb.chunk_id}")
    print(f"Type: {emb.type}")

Settings

Charting Settings

# Get charting settings
settings = client.charting_settings.get()
print(f"Primary Color: {settings.primary_color}")
print(f"Secondary Color: {settings.secondary_color}")

# Update charting settings
settings = client.charting_settings.update(
    primary_color="#FF5733",
    secondary_color="#33FF57",
    line_color="#3357FF"
)

Embeddings Settings

# Get embeddings settings
settings = client.embeddings_settings.get()
print(f"Max Chunk Words: {settings.max_chunk_words}")
print(f"Chunking Strategy: {settings.chunking_strategy}")

# Update embeddings settings
settings = client.embeddings_settings.update(
    max_chunk_words=1000,
    chunking_strategy="summary",
    image_extraction_strategy="safe"
)

Billing

Get Monthly Costs

costs = client.billing.get()
print(f"API Usage: {costs.api_usage}")
print(f"Training Usage: {costs.training_usage}")
print(f"Total Cost: ${costs.api_usage.get('totalCostUSD', 0):.2f}")

Streams

Get Output Stream

stream = client.streams.get(stream_id="stream-id")
print(f"Content: {stream.content}")
print(f"Type: {stream.type}")
print(f"Status: {stream.status}")

List Output Streams

streams = client.streams.list(limit=20)
for stream in streams.items:
    print(f"{stream.type} - {stream.status}")

Request Logs

Get Request Log

log = client.request_logs.get(log_id="log-id")
print(f"Type: {log.type}")
print(f"Status: {log.status}")
print(f"Tokens: {log.tokens}")
print(f"Words: {log.words}")

List Request Logs

logs = client.request_logs.list(limit=20)
for log in logs.items:
    print(f"{log.type} - {log.status} ({log.tokens} tokens)")

Streaming Responses

Stream responses with an iterator pattern, similar to OpenAI's Python SDK:

Simple Streaming

stream = client.streaming.send_to_agent(
    message="Tell me about your services",
    agent_id="your-agent-id"
)

for event in stream:
    print(event.text, end="", flush=True)

print()
print(f"Chat ID: {stream.chat_id}")
print(f"Message ID: {stream.message_id}")

Streaming with Event Types

stream = client.streaming.send_to_agent("Hello", "agent-id")

for event in stream:
    if event.is_token:
        # Token events contain streaming text
        print(event.text, end="", flush=True)
    elif event.is_complete:
        # Stream is complete
        print("\nDone!")
    elif event.is_error:
        # Handle errors
        print(f"Error: {event.data}")
    elif event.is_status:
        # Status updates
        print(f"Status: {event.data}")

Collect Full Response

stream = client.streaming.send_to_agent("Hello", "agent-id")
full_response = stream.collect()  # Blocks until complete
print(full_response)

Continue a Conversation

# First message creates a new chat
stream1 = client.streaming.send_to_agent("Hello", "agent-id")
for event in stream1:
    print(event.text, end="")

# Continue in the same chat
stream2 = client.streaming.send_to_agent(
    message="Tell me more",
    agent_id="agent-id",
    chat_id=stream1.chat_id  # Use the chat ID from first stream
)
for event in stream2:
    print(event.text, end="")

Streaming with Attachments

stream = client.streaming.send_to_agent(
    message="What's in this image?",
    agent_id="agent-id",
    attachments={
        "images": ["https://example.com/image.jpg"]
    }
)

for event in stream:
    print(event.text, end="")

Error Handling

from toothfairyai import ToothFairyClient, ToothFairyError

client = ToothFairyClient(api_key="...", workspace_id="...")

try:
    response = client.chat.send_to_agent("Hello", "agent-id")
except ToothFairyError as e:
    print(f"Error: {e.message}")
    print(f"Code: {e.code}")
    print(f"Status: {e.status_code}")
    if e.response:
        print(f"Response: {e.response}")

Specific Error Types

from toothfairyai import (
    ToothFairyError,
    ApiError,
    NetworkError,
    MissingApiKeyError,
    MissingWorkspaceIdError,
    ValidationError,
    FileSizeError
)

try:
    client = ToothFairyClient(api_key="", workspace_id="")
except MissingApiKeyError:
    print("API key is required")
except MissingWorkspaceIdError:
    print("Workspace ID is required")

try:
    result = client.documents.upload("large_file.pdf")
except FileSizeError as e:
    print(f"File too large: {e.size_mb:.2f}MB (max: {e.max_size_mb}MB)")
except ValidationError as e:
    print(f"Validation error: {e.message}")
except NetworkError as e:
    print(f"Network error: {e.message}")
except ApiError as e:
    print(f"API error: {e.message} (status: {e.status_code})")

Connection Testing

# Test connection
is_connected = client.test_connection()
print(f"Connected: {is_connected}")

# Get health status
health = client.get_health()
print(f"Status: {health.get('status', 'unknown')}")

Validation Models

The SDK includes optional Pydantic validation models to catch payload errors before making API requests. This is especially useful when integrating with AI agents (MCP tools) that need to construct payloads.

Why Use Validation Models?

• Catch errors early: Validate before sending to API
• Clear error messages: Know exactly what's wrong with your payload
• Required fields: Never forget required fields like interpolation_string (min 128 chars) for agents
• Type safety: Ensure correct types for all fields
• Enum validation: Prevent invalid enum values (e.g., CronJobFrequency.DAILY not daily)

Basic Usage

from toothfairyai import ToothFairyClient
from toothfairyai.models import AgentCreateModel, AgentFunctionCreateModel

client = ToothFairyClient(api_key="...", workspace_id="...")

# Validate before creating
agent_data = AgentCreateModel(
    label="Support Agent",
    mode="retriever",
    interpolation_string="You are a helpful customer support assistant. " * 10,  # Min 128 chars
    goals="Help customers resolve issues",
    temperature=0.7
)

# Convert to dict and pass to SDK
agent = client.agents.create(**agent_data.model_dump(exclude_none=True))

Validation with Error Handling

from toothfairyai.models import ScheduledJobCreateModel, ValidationError as ModelValidationError
from pydantic import ValidationError as PydanticValidationError

try:
    job_data = ScheduledJobCreateModel(
        name="Daily Report",
        agent_id="agent-123",
        custom_prompt_id="prompt-456",  # REQUIRED for scheduled jobs
        schedule={"frequency": "DAILY", "hour": 9, "minute": 0}
    )
except PydanticValidationError as e:
    print(f"Validation failed: {e}")
    # Shows exactly which fields are missing or invalid

Available Models

Model	Use Case	Key Requirements
AgentCreateModel	Create agents	interpolation_string ≥ 128 chars
AgentUpdateModel	Update agents	charting field required
AgentFunctionCreateModel	Create functions	description required
ScheduledJobCreateModel	Create scheduled jobs	custom_prompt_id required
DocumentCreateModel	Create documents	user_id required
ConnectionCreateModel	Create connections	name required
FolderCreateModel	Create folders	Uses name (not label)
EntityCreateModel	Create entities	background_color hex format
MemberCreateModel	Update members	Uses user_id (not email)
PromptCreateModel	Create prompts	interpolation_string required
AuthorisationCreateModel	Create auths	auth_type required
HookCreateModel	Create hooks	function_name required
BenchmarkCreateModel	Create benchmarks	questions list required
ChatCreateModel	Create chats	name required
ChannelCreateModel	Create channels	channel, provider required
SecretCreateModel	Create secrets	authorisation_id, password_secret_value required

Enum Types

from toothfairyai.models import CronJobFrequency, RequestType, AuthType

# CronJobFrequency uses UPPERCASE values
schedule = {"frequency": CronJobFrequency.DAILY, "hour": 9}

# RequestType for agent functions
function = AgentFunctionCreateModel(
    name="API Call",
    description="Fetch data from API",
    url="https://api.example.com",
    request_type=RequestType.GET  # Optional, defaults to None
)

# AuthType for authorisations
auth = AuthorisationCreateModel(
    name="API Key Auth",
    auth_type=AuthType.APIKEY
)

MCP Tool Integration

When building MCP tools or AI agents that create entities, use validation models to prevent payload errors:

def create_agent_tool(label: str, mode: str, interpolation_string: str, **kwargs):
    """MCP tool to create an agent with validation."""
    from toothfairyai.models import AgentCreateModel
    from pydantic import ValidationError
    
    try:
        # Validate the payload
        agent_data = AgentCreateModel(
            label=label,
            mode=mode,
            interpolation_string=interpolation_string,
            **kwargs
        )
        
        # If validation passes, create the agent
        client = get_client()
        return client.agents.create(**agent_data.model_dump(exclude_none=True))
        
    except ValidationError as e:
        # Return clear error message to AI
        return {"error": str(e), "validation_failed": True}

Integration with Existing Code

Validation models are optional and non-breaking. Existing code continues to work:

# Old code still works
agent = client.agents.create(
    label="Agent",
    mode="retriever",
    interpolation_string="..."  # Still works
)

# New code with validation
agent_data = AgentCreateModel(label="Agent", mode="retriever", interpolation_string="...")
agent = client.agents.create(**agent_data.model_dump(exclude_none=True))

Best Practices

1. Use Environment Variables for Credentials

import os
from toothfairyai import ToothFairyClient

client = ToothFairyClient(
    api_key=os.getenv("TOOTHFAIRY_API_KEY"),
    workspace_id=os.getenv("TOOTHFAIRY_WORKSPACE_ID")
)

2. Handle Pagination

def get_all_chats():
    all_chats = []
    offset = 0
    limit = 100

    while True:
        response = client.chat.list(limit=limit, offset=offset)
        all_chats.extend(response.items)
        
        if len(response.items) < limit:
            break
        
        offset += limit
    
    return all_chats

3. Use Streaming for Long Responses

# For long responses, use streaming to show progress
stream = client.streaming.send_to_agent(
    message="Generate a comprehensive report",
    agent_id="agent-id"
)

for event in stream:
    if event.is_token:
        print(event.text, end="", flush=True)

4. Implement Retry Logic

import time
from toothfairyai import NetworkError

def send_with_retry(message, agent_id, max_retries=3):
    for attempt in range(max_retries):
        try:
            return client.chat.send_to_agent(message, agent_id)
        except NetworkError as e:
            if attempt < max_retries - 1:
                time.sleep(2 ** attempt)  # Exponential backoff
            else:
                raise

5. Use Chat IDs for Conversations

# Create a chat once
chat = client.chat.create(name="Support Session")

# Use the same chat ID for all messages in the conversation
response1 = client.chat.send_to_agent("Hello", agent_id, chat_id=chat.id)
response2 = client.chat.send_to_agent("Tell me more", agent_id, chat_id=chat.id)

6. Organize Documents with Folders

# Create folder structure
procedures = client.folders.create(user_id="user-123", name="Procedures")
policies = client.folders.create(user_id="user-123", name="Policies")

# Upload documents to appropriate folders
client.documents.upload("procedure1.pdf", folder_id=procedures.id)
client.documents.upload("policy1.pdf", folder_id=policies.id)

7. Use Topics for Content Organization

# Create topics
support_topic = client.entities.create(
    user_id="user-123",
    label="FAQ Support",
    entity_type="topic"
)

# Associate documents with topics
doc = client.documents.create_from_path(
    file_path="user_guide.pdf",
    user_id="user-123",
    topics=[support_topic.id]
)

8. Monitor Usage with Billing

# Check monthly usage
costs = client.billing.get()
api_cost = costs.api_usage.get('totalCostUSD', 0)
training_cost = costs.training_usage.get('totalCharge', 0)

print(f"API Cost: ${api_cost:.2f}")
print(f"Training Cost: ${training_cost:.2f}")
print(f"Total: ${api_cost + training_cost:.2f}")

9. Use Scheduled Jobs for Automation

# Create a daily report job
job = client.scheduled_jobs.create(
    id="daily-report",
    name="Daily Summary Report",
    agent_id="report-agent-id",
    schedule={
        "frequency": "DAILY",
        "hour": 9,
        "minute": 0
    },
    timezone="UTC",
    is_active=True
)

10. Secure Secrets Management

# Create an authorisation for the external service
auth = client.authorisations.create(
    name="External Service Auth",
    auth_type="apikey",
    description="API key for external service integration"
)

# Store the actual secret value
secret = client.secrets.create(
    authorisation_id=auth.id,
    password_secret_value="your-actual-api-key"
)

# Reference the authorisation in agent functions
function = client.agent_functions.create(
    name="External API Call",
    description="Call external API with stored credentials",
    url="https://api.example.com/data",
    request_type="GET",
    authorisation_type="apikey",
    authorisation_key=auth.id  # Reference the authorisation
)

License

MIT License - see LICENSE for details.

Support

• Documentation: https://docs.toothfairyai.com
• Issues: GitHub Issues
• Email: [e-mail]

Changelog

See CHANGELOG.md for version history and updates.