Multi-Tenancy

Associate traces with specific tenants in multi-tenant applications.

What is Multi-Tenancy?

Multi-tenancy allows you to tag traces with a tenant identifier, enabling you to:

Filter traces by customer or organization
Track usage and costs per tenant
Debug issues for specific customers
Monitor performance across different tenants
Ensure data isolation in analytics

Setting the Tenant

Use set_tenant() to set the current tenant context:

1
2
3
4
5
6
7
8
9
10
11
import tracium
 
tracium.init(api_key="sk_live_...")
 
# Set the tenant for the current context
tracium.set_tenant("tenant-123")
 
# All traces now include tenant_id="tenant-123"
with tracium.start_trace(agent_name="my-agent") as trace:
    # This trace is associated with tenant-123
    pass

Getting the Current Tenant

Use get_current_tenant() to retrieve the current tenant:

1
2
3
4
5
6
7
8
9
10
11
12
import tracium
 
tracium.set_tenant("tenant-456")
 
# Get the current tenant
current = tracium.get_current_tenant()
print(current)  # "tenant-456"
 
# Returns None if no tenant is set
tracium.set_tenant(None)
current = tracium.get_current_tenant()
print(current)  # None

Web Framework Integration

Set the tenant at the start of each request based on authentication or headers:

FastAPI

fastapi_app.pypython

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
import tracium
from fastapi import FastAPI, Request, Depends
 
tracium.trace()
app = FastAPI()
 
def get_tenant_from_request(request: Request) -> str | None:
    """Extract tenant from JWT, header, or subdomain."""
    # From header
    tenant = request.headers.get("X-Tenant-ID")
    if tenant:
        return tenant
    
    # From JWT claims (example)
    # token = request.headers.get("Authorization")
    # claims = decode_jwt(token)
    # return claims.get("tenant_id")
    
    return None
 
@app.middleware("http")
async def tenant_middleware(request: Request, call_next):
    tenant = get_tenant_from_request(request)
    if tenant:
        tracium.set_tenant(tenant)
    
    response = await call_next(request)
    
    # Clear tenant after request
    tracium.set_tenant(None)
    return response
 
@app.post("/chat")
async def chat(message: str):
    # Traces here are associated with the tenant
    return {"response": "Hello!"}

Flask

flask_app.pypython

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
import tracium
from flask import Flask, request, g
 
tracium.trace()
app = Flask(__name__)
 
def get_tenant_from_request() -> str | None:
    """Extract tenant from request."""
    return request.headers.get("X-Tenant-ID")
 
@app.before_request
def set_tenant_context():
    tenant = get_tenant_from_request()
    if tenant:
        tracium.set_tenant(tenant)
        g.tenant_id = tenant
 
@app.teardown_request
def clear_tenant_context(exception=None):
    tracium.set_tenant(None)
 
@app.route("/chat", methods=["POST"])
def chat():
    # Traces here are associated with the tenant
    return {"response": "Hello!"}

Django

middleware.pypython

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
# middleware.py
import tracium
 
class TenantMiddleware:
    def __init__(self, get_response):
        self.get_response = get_response
 
    def __call__(self, request):
        # Extract tenant from request
        tenant = request.headers.get("X-Tenant-ID")
        
        if tenant:
            tracium.set_tenant(tenant)
        
        response = self.get_response(request)
        
        # Clear tenant after request
        tracium.set_tenant(None)
        
        return response
 
# settings.py
MIDDLEWARE = [
    # ... other middleware
    'your_app.middleware.TenantMiddleware',
]

Context Propagation

The tenant context automatically propagates through:

Async/await - Context variables propagate automatically
Threads - Tracium patches ThreadPoolExecutor for propagation
Background tasks - Works with Celery and other task queues

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
import tracium
import asyncio
 
tracium.init(api_key="sk_live_...")
 
async def process_request(tenant_id: str):
    tracium.set_tenant(tenant_id)
    
    # Tenant propagates to nested async calls
    await fetch_data()
    await generate_response()
 
async def fetch_data():
    # tenant_id is available here
    tenant = tracium.get_current_tenant()
    print(f"Fetching data for {tenant}")
 
async def generate_response():
    # tenant_id is still available
    tenant = tracium.get_current_tenant()
    print(f"Generating response for {tenant}")

Tenant ID Formats

The tenant ID can be any string up to 255 characters. Common formats include:

# UUID
tracium.set_tenant("550e8400-e29b-41d4-a716-446655440000")

# Slug
tracium.set_tenant("acme-corp")

# Prefixed ID
tracium.set_tenant("org_123456")
tracium.set_tenant("customer_abc123")

# Composite (org + workspace)
tracium.set_tenant("org_123:workspace_456")

Best Practices

Set Tenant Early

Set the tenant at the beginning of request handling, before any tracing occurs. Use middleware or decorators for consistency.

Clear Tenant After Requests

Always clear the tenant context after the request completes to prevent tenant leakage between requests.

Use Consistent IDs

Use the same tenant identifier across your system (database IDs, auth tokens, etc.) for easy correlation.

Handle Missing Tenants

Decide how to handle requests without a tenant - either reject them or use a default like "anonymous" or "system".

Complete Example

complete_example.pypython

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
import tracium
from fastapi import FastAPI, Request, HTTPException
from openai import OpenAI
 
# Initialize Tracium
tracium.init(
    api_key="sk_live_...",
    default_version="1.0.0",
)
tracium.trace()
 
app = FastAPI()
openai = OpenAI()
 
def extract_tenant(request: Request) -> str:
    """Extract and validate tenant from request."""
    tenant = request.headers.get("X-Tenant-ID")
    if not tenant:
        raise HTTPException(status_code=401, detail="Tenant ID required")
    return tenant
 
@app.middleware("http")
async def tenant_middleware(request: Request, call_next):
    try:
        tenant = extract_tenant(request)
        tracium.set_tenant(tenant)
    except HTTPException:
        # Allow the route to handle missing tenant
        pass
    
    response = await call_next(request)
    tracium.set_tenant(None)
    return response
 
@app.post("/chat")
async def chat(message: str, request: Request):
    tenant = tracium.get_current_tenant()
    if not tenant:
        raise HTTPException(status_code=401, detail="Tenant required")
    
    # This trace includes tenant_id automatically
    response = openai.chat.completions.create(
        model="gpt-4",
        messages=[{"role": "user", "content": message}]
    )
    
    return {
        "tenant": tenant,
        "response": response.choices[0].message.content
    }

← Versioning OpenAI Integration →