Tinybird Python Sdk Guidelines

Tinybird Python SDK Guidelines

Guidance for using the tinybird-sdk package to define Tinybird resources in Python.

When to Apply

• Installing or configuring tinybird-sdk
• Defining datasources, pipes, or endpoints in Python
• Creating Tinybird clients in Python
• Using data ingestion or queries in Python
• Running tinybird dev/build/deploy commands for Python projects
• Migrating from legacy .datasource/.pipe files to Python
• Defining connections (Kafka, S3, GCS)
• Creating materialized views, copy pipes, or sink pipes

Rule Files

• rules/getting-started.md
• rules/configuration.md
• rules/defining-datasources.md
• rules/defining-endpoints.md
• rules/client.md
• rules/low-level-api.md
• rules/cli-commands.md
• rules/connections.md
• rules/materialized-views.md
• rules/copy-sink-pipes.md
• rules/tokens.md

Quick Reference

• Install: pip install tinybird-sdk
• Initialize: tinybird init
• Dev mode: tinybird dev (uses configured dev_mode, typically branch)
• Build: tinybird build (builds against configured dev target)
• Deploy: tinybird deploy (deploys to main/production)
• Preview in CI: tinybird preview
• Migrate: tinybird migrate (convert .datasource/.pipe files to Python)
• Server-side only; never expose tokens in browsers

SDK CLI Commands

The SDK installs tinybird as a runtime dependency. Some commands are handled by the SDK; others delegate to the Tinybird CLI.

CLI 4.0 Build/Deploy Model

• Configure your default development target once in tinybird.config.* (dev_mode).
• Run tinybird build without environment flags for normal workflows.
• Run tinybird deploy to publish to Tinybird Cloud main.
• Use --local/--branch only as explicit overrides.

tinybird init

Initialize a new Tinybird project:

tinybird init
tinybird init --force          # Overwrite existing files
tinybird init --skip-login     # Skip browser authentication

Creates lib/datasources.py, lib/pipes.py, lib/client.py, and tinybird.config.json.

tinybird migrate

Migrate legacy datafiles to Python definitions:

tinybird migrate "tinybird/**/*.datasource" "tinybird/**/*.pipe" "tinybird/**/*.connection"
tinybird migrate tinybird/legacy --out ./tinybird.migration.py
tinybird migrate tinybird --dry-run

Converts .datasource, .pipe, and .connection files into a Python definitions file.

tinybird dev

Watch schema files and auto-sync to Tinybird:

tinybird dev                   # Watch and sync using configured dev_mode
tinybird dev --local           # Sync with local container
tinybird dev --branch          # Force branch mode for this run

Important: In branch mode, feature branches are expected; main/master are blocked to prevent accidental production changes.

tinybird build

Build and validate resources using your configured development target:

tinybird build                 # Build to dev_mode target (branch or local)
tinybird build --dry-run       # Preview build operations
tinybird build --local         # Build to local container
tinybird build --branch        # Build to branch for this run

Use tinybird build for iterative development; it does not publish to production.

tinybird deploy

Deploy resources to the main workspace (production):

tinybird deploy                # Deploy to main/production
tinybird deploy --dry-run      # Preview without deploying
tinybird deploy --check        # Validate without deploying
tinybird deploy --wait         # Wait for deployment completion
tinybird deploy --allow-destructive-operations  # Allow breaking changes

This is the only way to deploy to main.

tinybird preview

Create or refresh a CI preview environment for the current branch:

tinybird preview

Use this in pull request workflows so preview apps query isolated Tinybird preview branches.

tinybird pull

Pull resources from remote workspace:

tinybird pull                  # Pull to default location
tinybird pull --output-dir ./tinybird-datafiles
tinybird pull --force          # Overwrite existing files

tinybird login

Authenticate via browser:

tinybird login

Useful for existing projects or token refresh.

tinybird branch

Manage branches:

tinybird branch list           # List all branches
tinybird branch status         # Show current branch status
tinybird branch delete <name>  # Delete a branch

tinybird info

Display workspace, local, and project configuration:

tinybird info                  # Show configuration
tinybird info --json           # Output as JSON

Development Workflow

1. tinybird init - Initialize project 2. Define datasources and pipes in Python 3. tinybird build or tinybird dev - Iterate against configured dev target 4. tinybird preview in CI - Create preview branch environment per PR 5. tinybird deploy - Deploy to production after merge

Migration Workflow

1. tinybird migrate "path/to/*.datasource" "path/to/*.pipe" - Convert legacy files 2. Review generated Python file 3. Move definitions to lib/datasources.py and lib/pipes.py 4. Update tinybird.config.json to include Python files 5. tinybird dev - Verify sync works

Important Notes

• The CLI auto-generates datafiles from Python definitions before build, deploy, and preview
• Use --check/--dry-run before production deploys when in doubt
• The CLI automatically loads .env.local and .env files

Creating the Tinybird Client

Client Setup

# lib/client.py
from tinybird_sdk import Tinybird
from .datasources import page_views
from .pipes import top_pages

tinybird = Tinybird(
    {
        "datasources": {"page_views": page_views},
        "pipes": {"top_pages": top_pages},
    }
)

__all__ = ["tinybird", "page_views", "top_pages"]

Using the Client

Data Ingestion

from lib.client import tinybird

# Ingest one row
tinybird.page_views.ingest(
    {
        "timestamp": "2024-01-15 10:30:00",
        "pathname": "/home",
        "session_id": "abc123",
        "country": "US",
    }
)

# Batch ingestion (list of rows)
tinybird.page_views.ingest([
    {"timestamp": "2024-01-15 10:30:00", "pathname": "/home", "session_id": "abc", "country": "US"},
    {"timestamp": "2024-01-15 10:31:00", "pathname": "/about", "session_id": "abc", "country": "US"},
])

Querying Endpoints

from lib.client import tinybird

result = tinybird.top_pages.query(
    {
        "start_date": "2024-01-01 00:00:00",
        "end_date": "2024-01-31 23:59:59",
        "limit": 5,
    }
)

# Access result data
for row in result["data"]:
    print(f"{row['pathname']}: {row['views']} views")

Datasource Operations

The client provides several operations for managing datasource data:

Append from URL

tinybird.page_views.append(
    {
        "url": "https://example.com/page_views.csv",
    }
)

Replace (Full Snapshot)

tinybird.page_views.replace(
    {
        "url": "https://example.com/page_views_full_snapshot.csv",
    }
)

Delete Rows

# Delete matching rows
tinybird.page_views.delete(
    {
        "delete_condition": "country = 'XX'",
    }
)

# Dry run to preview deletions
tinybird.page_views.delete(
    {
        "delete_condition": "country = 'XX'",
        "dry_run": True,
    }
)

Truncate

tinybird.page_views.truncate()

Client Benefits

• Convenience: Access datasources and pipes as attributes
• Consistency: All operations use the same pattern
• Organization: Keep definitions and client in dedicated modules

Python App Integration

For Python web apps (FastAPI, Django, Flask), import from a dedicated module:

# In your FastAPI app
from lib.client import tinybird

@app.get("/analytics")
async def get_analytics():
    result = tinybird.top_pages.query({"start_date": "2024-01-01", "end_date": "2024-01-31"})
    return result["data"]

SDK Configuration

Configuration File

Create a configuration file in your project root. Supported formats (in priority order):

1. tinybird.config.py - Python config with dynamic logic 2. tinybird_config.py - Python config alias 3. tinybird.config.json - Standard JSON (default) 4. tinybird.json - Legacy format

JSON Configuration

{
  "include": [
    "lib/*.py",
    "tinybird/**/*.datasource",
    "tinybird/**/*.pipe",
    "tinybird/**/*.connection"
  ],
  "token": "${TINYBIRD_TOKEN}",
  "base_url": "https://api.tinybird.co",
  "dev_mode": "branch"
}

Python Configuration

# tinybird.config.py
config = {
    "include": ["lib/*.py"],
    "token": "${TINYBIRD_TOKEN}",
    "base_url": "https://api.tinybird.co",
    "dev_mode": "branch",
}

For Python configs, export one of:

• config dict
• CONFIG dict
• default dict
• get_config() returning a dict

Configuration Options

Token Resolution

If token is omitted, SDK resolves from: 1. TINYBIRD_TOKEN environment variable 2. .tinyb file

Base URL Resolution

If base_url is omitted, SDK resolves from: 1. TINYBIRD_URL environment variable 2. TINYBIRD_HOST environment variable 3. .tinyb file (host field) 4. Default: https://api.tinybird.co

Mixed Formats

Combine Python files with legacy .datasource, .pipe, and .connection files:

{
  "include": [
    "lib/datasources.py",
    "lib/pipes.py",
    "legacy/events.datasource",
    "legacy/analytics.pipe"
  ]
}

Local Development Mode

Use a local Tinybird container:

1. Start the container:

   docker run -d -p 7181:7181 --name tinybird-local tinybirdco/tinybird-local:latest

2. Configure your project:

   {
     "dev_mode": "local"
   }

Or use CLI flag:

   tinybird dev --local

Defining Connections

Connections define external data sources that Tinybird can integrate with.

Kafka Connection

from tinybird_sdk import define_kafka_connection, secret

events_kafka = define_kafka_connection(
    "events_kafka",
    {
        "bootstrap_servers": "kafka.example.com:9092",
        "security_protocol": "SASL_SSL",
        "sasl_mechanism": "PLAIN",
        "key": secret("KAFKA_KEY"),
        "secret": secret("KAFKA_SECRET"),
    },
)

S3 Connection

from tinybird_sdk import define_s3_connection

landing_s3 = define_s3_connection(
    "landing_s3",
    {
        "region": "us-east-1",
        "arn": "arn:aws:iam::123456789012:role/tinybird-s3-access",
    },
)

GCS Connection

from tinybird_sdk import define_gcs_connection, secret

landing_gcs = define_gcs_connection(
    "landing_gcs",
    {
        "service_account_credentials_json": secret("GCS_SERVICE_ACCOUNT_CREDENTIALS_JSON"),
    },
)

Using Secrets

The secret() function references secrets stored in Tinybird:

from tinybird_sdk import secret

# Reference a secret by name
api_key = secret("MY_API_KEY")

Secrets must be created in Tinybird before deploying connections that use them.

Connection Configuration Options

Kafka Options

S3 Options

GCS Options

Using Connections in Sink Pipes

Connections are referenced when defining sink pipes:

from tinybird_sdk import define_sink_pipe, node

kafka_sink = define_sink_pipe(
    "kafka_events_sink",
    {
        "sink": {
            "connection": events_kafka,  # Reference the connection
            "topic": "events_export",
            "schedule": "@on-demand",
        },
        "nodes": [
            node({"name": "publish", "sql": "SELECT * FROM events"})
        ],
    },
)

Copy Pipes and Sink Pipes

Copy Pipes

Copy pipes execute SQL and write results to a datasource on a schedule or on-demand.

Scheduled Copy Pipe

from tinybird_sdk import define_copy_pipe, node

daily_snapshot = define_copy_pipe(
    "daily_snapshot",
    {
        "datasource": events,  # Target datasource
        "copy_schedule": "0 0 * * *",  # Cron: daily at midnight
        "copy_mode": "append",
        "nodes": [
            node(
                {
                    "name": "snapshot",
                    "sql": """
                        SELECT today() AS snapshot_date, event_name, count() AS events
                        FROM events
                        WHERE toDate(timestamp) = today() - 1
                        GROUP BY event_name
                    """,
                }
            )
        ],
    },
)

On-Demand Copy Pipe

manual_report = define_copy_pipe(
    "manual_report",
    {
        "datasource": events,
        "copy_schedule": "@on-demand",
        "copy_mode": "replace",
        "nodes": [
            node(
                {
                    "name": "report",
                    "sql": "SELECT * FROM events WHERE timestamp >= now() - interval 7 day",
                }
            )
        ],
    },
)

Copy Modes

Schedule Options

Sink Pipes

Sink pipes publish query results to external systems (Kafka, S3).

Kafka Sink

from tinybird_sdk import define_sink_pipe, node

kafka_events_sink = define_sink_pipe(
    "kafka_events_sink",
    {
        "sink": {
            "connection": events_kafka,  # Kafka connection
            "topic": "events_export",
            "schedule": "@on-demand",
        },
        "nodes": [
            node(
                {
                    "name": "publish",
                    "sql": "SELECT timestamp, payload FROM kafka_events",
                }
            )
        ],
    },
)

S3 Sink

s3_events_sink = define_sink_pipe(
    "s3_events_sink",
    {
        "sink": {
            "connection": landing_s3,  # S3 connection
            "bucket_uri": "s3://my-bucket/exports/",
            "file_template": "events_{date}",
            "format": "csv",
            "schedule": "@once",
            "strategy": "create_new",
            "compression": "gzip",
        },
        "nodes": [
            node(
                {
                    "name": "export",
                    "sql": "SELECT timestamp, session_id FROM s3_landing",
                }
            )
        ],
    },
)

S3 Sink Options

Defining Datasources

Basic Datasource Definition

from tinybird_sdk import define_datasource, t, engine

page_views = define_datasource(
    "page_views",
    {
        "description": "Page view tracking data",
        "schema": {
            "timestamp": t.date_time(),
            "pathname": t.string(),
            "session_id": t.string(),
            "country": t.string().low_cardinality().nullable(),
        },
        "engine": engine.merge_tree(
            {
                "sorting_key": ["pathname", "timestamp"],
            }
        ),
    },
)

Schema Types

The t object provides type definitions:

String Types

• t.string() - Variable-length string
• t.fixed_string(n) - Fixed-length string
• t.uuid() - UUID type

Numeric Types

• t.int32(), t.int64() - Signed integers
• t.uint32(), t.uint64() - Unsigned integers
• t.float32(), t.float64() - Floating point
• t.decimal(precision, scale) - Decimal type

Date/Time Types

• t.date_time() - DateTime type
• t.date_time64(precision) - DateTime64 with precision (0-9)
• t.date() - Date type

Other Types

• t.bool() - Boolean type (stored as UInt8)
• t.array(inner_type) - Array of any type
• t.map(key_type, value_type) - Map/dictionary type

Aggregate Types

• t.simple_aggregate_function(func, inner_type) - For summing merge tree
• t.aggregate_function(func, inner_type) - For aggregating merge tree

Type Modifiers

Chain modifiers on types:

• .nullable() - Make column nullable
• .low_cardinality() - Use LowCardinality encoding for low-unique strings
• .default(value) - Set default value

Example:

schema = {
    "tags": t.array(t.string()),
    "country": t.string().low_cardinality().nullable(),
    "score": t.float64().nullable(),
    "status": t.string().default("pending"),
}

Engine Configuration

MergeTree

engine.merge_tree(
    {
        "sorting_key": ["column1", "column2"],
        "partition_key": "toYYYYMM(timestamp)",  # optional
        "ttl": "timestamp + INTERVAL 90 DAY",   # optional
    }
)

ReplacingMergeTree

engine.replacing_merge_tree(
    {
        "sorting_key": ["id"],
        "ver": "updated_at",
    }
)

SummingMergeTree

engine.summing_merge_tree(
    {
        "sorting_key": ["date", "category"],
        "columns": ["count", "total"],
    }
)

AggregatingMergeTree

engine.aggregating_merge_tree(
    {
        "sorting_key": ["date", "dimension"],
    }
)

Schema Inference

Use the infer module to extract schemas:

from tinybird_sdk.infer import infer_row_schema

row_schema = infer_row_schema(page_views)
# Returns dict with column names and types

Defining Endpoints (Pipes)

Basic Endpoint Definition

from tinybird_sdk import define_endpoint, node, t, p

top_pages = define_endpoint(
    "top_pages",
    {
        "description": "Get the most visited pages",
        "params": {
            "start_date": p.date_time(),
            "end_date": p.date_time(),
            "limit": p.int32().optional(10),
        },
        "nodes": [
            node(
                {
                    "name": "aggregated",
                    "sql": """
                        SELECT pathname, count() AS views
                        FROM page_views
                        WHERE timestamp >= {{DateTime(start_date)}}
                          AND timestamp <= {{DateTime(end_date)}}
                        GROUP BY pathname
                        ORDER BY views DESC
                        LIMIT {{Int32(limit, 10)}}
                    """,
                }
            )
        ],
        "output": {
            "pathname": t.string(),
            "views": t.uint64(),
        },
    },
)

Parameter Types

The p object provides parameter definitions:

• p.string() - String parameter
• p.int32(), p.int64() - Integer parameters
• p.float32(), p.float64() - Float parameters
• p.date_time() - DateTime parameter
• p.date() - Date parameter

Parameter Modifiers

• .optional(default_value) - Make parameter optional with a default
• .describe(text) - Add description for documentation

Example:

params = {
    "limit": p.int32().optional(10),
    "filter": p.string().optional(""),
    "status": p.string().optional("active").describe("Filter by status"),
}

Internal Pipes (Non-API)

Use define_pipe for pipes not exposed as API endpoints:

from tinybird_sdk import define_pipe, node, p

filtered_events = define_pipe(
    "filtered_events",
    {
        "description": "Filter events by date range",
        "params": {
            "start_date": p.date_time(),
            "end_date": p.date_time(),
        },
        "nodes": [
            node(
                {
                    "name": "filtered",
                    "sql": """
                        SELECT * FROM events
                        WHERE timestamp >= {{DateTime(start_date)}}
                          AND timestamp <= {{DateTime(end_date)}}
                    """,
                }
            )
        ],
    },
)

Multi-Node Pipes

Define multiple nodes for complex transformations:

nodes = [
    node(
        {
            "name": "filtered",
            "sql": """
                SELECT * FROM events
                WHERE timestamp >= {{DateTime(start_date)}}
            """,
        }
    ),
    node(
        {
            "name": "aggregated",
            "sql": """
                SELECT date, count() as total
                FROM filtered
                GROUP BY date
            """,
        }
    ),
]

SQL Templating

Use Tinybird templating in SQL:

• {{Type(param_name)}} - Parameter with type
• {{Type(param_name, default)}} - Parameter with default value

WHERE user_id = {{String(user_id)}}
AND date >= {{Date(start_date, '2024-01-01')}}
LIMIT {{Int32(limit, 100)}}

Schema Inference

from tinybird_sdk.infer import infer_params_schema, infer_output_schema

params_schema = infer_params_schema(top_pages)
output_schema = infer_output_schema(top_pages)

Tinybird Python SDK Overview

What is it

The tinybird-sdk is a Python package that enables developers to define Tinybird resources in Python. You can author datasources, pipes, connections, and queries in Python, then synchronize them directly to Tinybird.

Requirements

• Python: Version 3.11 or higher
• Server-side only; web browsers are not supported to protect API credentials

Installation

pip install tinybird-sdk

Project Initialization

tinybird init
tinybird init --force          # Overwrite existing files
tinybird init --skip-login     # Skip browser authentication

This generates:

• tinybird.config.json - Configuration file
• lib/datasources.py - Data source definitions
• lib/pipes.py - Pipe/endpoint definitions
• lib/client.py - Tinybird client module

Environment Setup

Create .env.local:

TINYBIRD_TOKEN=p.your_token_here

Key Features

• Define datasources, pipes, and endpoints in Python
• Data ingestion with automatic schema validation
• Query endpoints with typed results
• Mixed formats: combine Python with legacy .datasource/.pipe files
• Branch safety: dev mode blocks deployment to main branch
• Connections: Kafka, S3, GCS integrations
• Materialized views for real-time aggregations
• Copy pipes and sink pipes for data workflows

Public Tinybird API (Low-Level)

For cases requiring a decoupled API wrapper without the high-level client:

Creating the API Client

from tinybird_sdk import create_tinybird_api

api = create_tinybird_api(
    {
        "base_url": "https://api.tinybird.co",
        "token": "p.your_token",
    }
)

Querying Endpoints

top_pages = api.query(
    "top_pages",
    {
        "start_date": "2024-01-01",
        "end_date": "2024-01-31",
        "limit": 5,
    },
)

# top_pages["data"] contains the result rows

Ingesting Data

# Ingest one row
api.ingest(
    "events",
    {
        "timestamp": "2024-01-15 10:30:00",
        "event_name": "page_view",
        "pathname": "/home",
    },
)

# Batch ingestion
api.ingest(
    "events",
    [
        {"timestamp": "2024-01-15 10:30:00", "event_name": "page_view", "pathname": "/home"},
        {"timestamp": "2024-01-15 10:31:00", "event_name": "click", "pathname": "/home"},
    ],
)

Retry Behavior

Retries are disabled by default. Enable with max_retries:

api.ingest(
    "events",
    {"timestamp": "2024-01-15 10:31:00", "event_name": "button_click", "pathname": "/pricing"},
    {"max_retries": 3},
)

• 429 retries use Retry-After / X-RateLimit-Reset headers
• 503 retries use SDK default exponential backoff

Datasource Operations

Append from URL

api.append_datasource(
    "events",
    {"url": "https://example.com/events.csv"},
)

Delete Rows

api.delete_datasource(
    "events",
    {"delete_condition": "event_name = 'test'"},
)

# Dry run
api.delete_datasource(
    "events",
    {"delete_condition": "event_name = 'test'", "dry_run": True},
)

Truncate

api.truncate_datasource("events")

Executing Raw SQL

sql_result = api.sql("SELECT count() AS total FROM events")
# sql_result["data"][0]["total"]

Per-Request Token Override

workspace_response = api.request_json(
    "/v1/workspace",
    token="p.branch_or_jwt_token",
)

When to Use Low-Level API

• Existing projects not using Python definitions
• Dynamic endpoint names or parameters
• Direct SQL execution needs
• Gradual migration from other HTTP clients
• Multi-tenant scenarios with different tokens

Materialized Views

Materialized views automatically aggregate data as it arrives, enabling real-time analytics.

Basic Materialized View

A materialized view consists of: 1. A target datasource with aggregate columns 2. A materialized view definition that populates it

from tinybird_sdk import define_datasource, define_materialized_view, engine, node, t

# Target datasource with aggregate columns
daily_stats = define_datasource(
    "daily_stats",
    {
        "schema": {
            "date": t.date(),
            "pathname": t.string(),
            "views": t.simple_aggregate_function("sum", t.uint64()),
            "unique_sessions": t.aggregate_function("uniq", t.string()),
        },
        "engine": engine.aggregating_merge_tree({"sorting_key": ["date", "pathname"]}),
    },
)

# Materialized view that populates it
daily_stats_mv = define_materialized_view(
    "daily_stats_mv",
    {
        "datasource": daily_stats,
        "nodes": [
            node(
                {
                    "name": "aggregate",
                    "sql": """
                        SELECT
                          toDate(timestamp) AS date,
                          pathname,
                          count() AS views,
                          uniqState(session_id) AS unique_sessions
                        FROM page_views
                        GROUP BY date, pathname
                    """,
                }
            )
        ],
    },
)

Aggregate Types

SimpleAggregateFunction

For simple aggregations (sum, min, max, any):

"views": t.simple_aggregate_function("sum", t.uint64())
"min_value": t.simple_aggregate_function("min", t.float64())
"max_value": t.simple_aggregate_function("max", t.float64())

AggregateFunction

For complex aggregations (uniq, quantile, etc.):

"unique_users": t.aggregate_function("uniq", t.string())
"p95_latency": t.aggregate_function("quantile(0.95)", t.float64())

SQL State Functions

In materialized view SQL, use state functions to prepare aggregates:

Querying Materialized Views

When querying, use merge functions for AggregateFunction columns:

endpoint = define_endpoint(
    "daily_stats_query",
    {
        "nodes": [
            node(
                {
                    "name": "query",
                    "sql": """
                        SELECT
                          date,
                          pathname,
                          sum(views) AS total_views,
                          uniqMerge(unique_sessions) AS unique_sessions
                        FROM daily_stats
                        GROUP BY date, pathname
                    """,
                }
            )
        ],
        "output": {
            "date": t.date(),
            "pathname": t.string(),
            "total_views": t.uint64(),
            "unique_sessions": t.uint64(),
        },
    },
)

Engine Selection

Always use aggregating_merge_tree for materialized view targets:

engine.aggregating_merge_tree(
    {
        "sorting_key": ["date", "dimension1", "dimension2"],
    }
)

Tokens

Static Tokens

Define named tokens and attach them to datasources and endpoints:

from tinybird_sdk import define_datasource, define_endpoint, define_token, node, t

# Define tokens
app_token = define_token("app_read")
ingest_token = define_token("ingest_token")

# Attach to datasource
events = define_datasource(
    "events",
    {
        "schema": {
            "timestamp": t.date_time(),
            "event_name": t.string(),
        },
        "tokens": [
            {"token": app_token, "scope": "READ"},
            {"token": ingest_token, "scope": "APPEND"},
        ],
    },
)

# Attach to endpoint
top_events = define_endpoint(
    "top_events",
    {
        "nodes": [node({"name": "endpoint", "sql": "SELECT * FROM events LIMIT 10"})],
        "output": {"timestamp": t.date_time(), "event_name": t.string()},
        "tokens": [{"token": app_token, "scope": "READ"}],
    },
)

Token Scopes

JWT Token Creation

Create short-lived JWT tokens for secure scoped access:

from datetime import datetime, timedelta, timezone
from tinybird_sdk import create_client

client = create_client(
    {
        "base_url": "https://api.tinybird.co",
        "token": "p.your_admin_token",
    }
)

result = client.tokens.create_jwt(
    {
        "name": "user_123_session",
        "expires_at": datetime.now(tz=timezone.utc) + timedelta(hours=1),
        "scopes": [
            {
                "type": "PIPES:READ",
                "resource": "user_dashboard",
                "fixed_params": {"user_id": 123},
            }
        ],
        "limits": {"rps": 10},
    }
)

jwt_token = result["token"]

JWT Scope Types

JWT Scope Options

Example: Multi-Tenant Access

# Create token for specific organization
org_token = client.tokens.create_jwt(
    {
        "name": "org_acme_access",
        "expires_at": datetime.now(tz=timezone.utc) + timedelta(days=1),
        "scopes": [
            {
                "type": "DATASOURCES:READ",
                "resource": "events",
                "filter": "org_id = 'acme'",
            },
            {
                "type": "PIPES:READ",
                "resource": "analytics_dashboard",
                "fixed_params": {"org_id": "acme"},
            },
        ],
        "limits": {"rps": 100},
    }
)

JWT Limits