fastn SDK for Python

Embedded integrations for your product and your AI agents.

fastn is the embedded integration platform for B2B SaaS. fastn-ai is its official Python SDK: add customer-facing integrations (CRM, calendar, Slack, and 1,000+ more apps) to your product through one API, and give your AI agents the same integrations as tools.

Your customers connect their own apps inside your product. Your backend uses those connections through the fastn platform API (fastn.platform), and your AI agents use them as tools with OpenAI, Anthropic, Gemini, or Bedrock function calling.

Built for multi-tenant SaaS:

• One key, every customer. Address each customer by your own id (fastn.customer("acme-42")); fastn's internal org ids never reach your code. Every call is scoped to that customer, and an account or watch id that belongs to another customer is refused, so one customer's data never leaks into another's.
• Each customer brings their own accounts. Send a setup link and the customer connects their HubSpot, Google Calendar, Slack, and so on. Credentials stay in fastn's vault, isolated per customer, and refresh automatically.
• Several accounts per app. A customer can connect two Slack workspaces or two calendars; pick one per call with connection_id.
• One API across providers. customer.crm and customer.calendar work the same whichever CRM or calendar the customer connected; the provider is picked from their connections.
• Safe retries and traceability. idempotency_key makes writes safe to retry, and correlation_id lets you ask later what fastn did for a request (customer.call_log(...)).
• Governed. Role-based access control, audit logs, and per-customer isolation on a SOC 2 Type II platform.

Use fastn when:

• your customers ask for integrations your team cannot build and maintain one by one;
• every customer needs the same integration connected to their own account and configured to their own fields;
• you want one integration to serve your product, your customers, and your AI agents;
• you need per-customer isolation, audit logs, and SOC 2 for a security review.

Which client do I need?

The package ships two clients. They use different keys and endpoints, so pick by what you are building:

Embed fastn in your product (guide for Claude and other AI coding assistants)

The steps to add customer-facing integrations to a SaaS app with fastn. Use the exact names below; do not invent props or endpoints.

1. Install and get a key. pip install fastn-ai, then create a platform API key in the fastn dashboard and set it as FASTN_API_KEY.

2. Register each of your customers by your own id (backend):

   import os
   from fastn.platform import Apps, FastnClient
   
   fastn = FastnClient(api_key=os.environ["FASTN_API_KEY"])
   customer = fastn.customer(account.id)          # your id for this customer
   customer.register({"name": account.name})      # idempotent
   

3. Let the customer connect their apps. Either send a hosted setup link, or show the embedded widget in your UI:

   link = customer.connect(Apps.hubspot)             # hosted page: redirect the customer to link["url"]
   token = customer.embed_token()                 # or: a short-lived token for the in-app widget
   

   On the frontend, the widget is the @fastn-ai/embed package (FastnProvider with FastnHub or FastnConnectCard from @fastn-ai/embed/react, or the hosted script tag). Copy the exact snippet from the dashboard's Embed tab. See the embedding docs.

4. Use the customer's connections from your backend: customer.crm.contacts.find(...), customer.calendar.book(...), or customer.run_action(app, action, input) for any app action.

5. React to changes: fastn.webhooks.create(...) once for your account, customer.watch(app, events) per customer, and verify deliveries with parse_event(secret, headers, body).

6. Give your AI agents the same integrations: use get_tools_for() below, or connect Claude, Cursor, or any MCP client to the fastn MCP gateway.

Pricing is per customer connection, with a free plan to start: see fastn.ai/pricing. A machine-readable summary of fastn for LLMs is at fastn.ai/llms.txt.

Why fastn?

Installation

pip install "fastn-ai[cli]"   # SDK + the fastn CLI (recommended)
pip install fastn-ai          # SDK only, no CLI

Requires Python 3.8+. The package includes type stubs for 250+ connectors, so IDE autocomplete works out of the box. The fastn command needs the [cli] extra.

Quick Start

Log in once and add the connectors you want to call. This writes your credentials and the connector tool schemas to .fastn/ in the current directory:

fastn login                  # opens the browser, then pick a project
fastn connector sync         # fetch the connector registry
fastn connector add slack    # fetch Slack's tool schemas

Then call the connector from Python:

from fastn import FastnClient

fastn = FastnClient()        # reads .fastn/config.json
fastn.slack.send_message(channel="general", text="Hello from fastn!")

Calling a connector that has not been added raises ConnectorNotFoundError. On a server, pass api_key= and project_id= (or set FASTN_API_KEY and FASTN_PROJECT_ID) instead of fastn login. get_tools_for() needs no local registry.

Async:

from fastn import AsyncFastnClient

async with AsyncFastnClient() as fastn:
    await fastn.slack.send_message(channel="general", text="Hello!")

How It Works

fastn sits between your AI agent and the apps your customers connect. The SDK handles client-side concerns (tool discovery, schema transformation, parameter routing). The platform handles server-side concerns (credential injection, access control, batch optimization, prompt safety, observability).

One-liner: Connectors provide tools. Flows compose tools. Agents run tools and flows with reasoning.

Core Capabilities

Dynamic Tool Filtering

get_tools_for(prompt) sends a semantic search to the platform, which matches the prompt against the full connector registry and returns only the top N tools (default: 5). Sending a handful of tools instead of hundreds keeps the tool context small, and fewer tools also means fewer hallucinated tool calls.

from fastn import Connectors, ToolFormats
# Semantic discovery: only the most relevant tools reach the LLM
tools = fastn.get_tools_for("Send a Slack message", format=ToolFormats.openai, limit=5)

# Direct lookup: bypass semantic search when you know the connector name
tools = fastn.get_tools_for("slack tools", connector=Connectors.slack, format=ToolFormats.openai)

Context Optimization

fastn minimizes what reaches the LLM: fewer tokens, less hallucination, better compliance.

• Tool and skill composition: the platform composes tools and skills into efficient chains, so agents receive focused context instead of raw API surface area
• Schema I/O filtering: strips unnecessary fields from tool input and output schemas before they reach the LLM, reducing token cost
• Schema transformation: the SDK flattens nested API wrappers (body, param) for the LLM and re-wraps them for execution:

What the API expects:   {"body": {"channel": "#general", "text": "hello"}}
What the LLM sees:      {"channel": "#general", "text": "hello"}

Combined with Dynamic Tool Filtering, the agent gets a focused payload instead of the full API surface.

Architecture

┌─────────────────────────────────────────────────────────────────────────┐
│  Your Agent                                                             │
│                                                                         │
│  1. Discovery                                                           │
│     get_tools_for(prompt) ──→ SDK ──→ POST /getTools ──→ Platform       │
│     Platform: dynamic tool filtering, intent routing                    │
│                              ←── top N tools (unwrapped schemas) ←──    │
│                                                                         │
│  2. LLM Call                                                            │
│     SDK: format schemas for provider (OpenAI / Claude / Gemini / etc.)  │
│     Send prompt + N tool schemas to LLM                                 │
│     LLM returns tool_call with flat params                              │
│                                                                         │
│  3. Execution                                                           │
│     execute(tool, params) ──→ SDK (re-wrap params) ──→ Platform         │
│     Platform: credential injection, RBAC, prompt safety, batch opt.     │
│                               ←── result ←──                            │
│                                                                         │
│  4. Observability                                                       │
│     Platform: audit trail, cost/latency/error tracking per tool         │
│     SDK CLI: per-call cost, latency, and token summary table            │
└─────────────────────────────────────────────────────────────────────────┘

Terminology

SDKs

Configuration

Credentials are loaded from three sources (highest priority first):

1. Constructor parameters: FastnClient(api_key="...", project_id="...")
2. Environment variables: FASTN_API_KEY, FASTN_PROJECT_ID
3. Config file: .fastn/config.json (created by fastn login)

# Explicit credentials (no config file needed)
fastn = FastnClient(api_key="your-api-key", project_id="your-project-id")

LLM Agent Integration

fastn handles the three hardest parts of giving LLMs access to tools:

1. Discovery: get_tools_for(prompt) semantically matches tools from 250+ connectors. Only the top N reach the LLM, keeping context small and hallucination low.
2. Schema translation: schemas are flattened for LLM consumption and formatted for your provider (OpenAI, Anthropic, Gemini, Bedrock).
3. Execution: execute(tool, params) routes through the platform, which handles credential injection, parameter re-wrapping, retries, and logging.

import json
import openai
from fastn import FastnClient, ToolFormats

fastn = FastnClient()

# 1. Describe what you need and fastn discovers the right tools
tools = fastn.get_tools_for(
    "Send a message on Slack and create a Jira ticket",
    format=ToolFormats.openai,   # also: anthropic, gemini, bedrock, raw
)

# 2. Send tools + user prompt to the LLM
response = openai.chat.completions.create(
    model="gpt-4o",
    messages=[{"role": "user", "content": "Send hello to #general on Slack"}],
    tools=tools,
)

# 3. The LLM returns a tool_call when it wants to use a tool
tool_call = response.choices[0].message.tool_calls[0]

# 4. Execute the tool call through fastn
result = fastn.execute(
    tool=tool_call.function.name,                      # e.g. "send_message"
    params=json.loads(tool_call.function.arguments),    # e.g. {"channel": "general", "text": "hello"}
)
print(result)

Supported LLM Providers

Anthropic Claude Example

import anthropic
from fastn import FastnClient, ToolFormats

fastn = FastnClient()
tools = fastn.get_tools_for("Send a message on Slack", format=ToolFormats.anthropic)

client = anthropic.Anthropic()
response = client.messages.create(
    model="claude-sonnet-4-20250514",
    max_tokens=1024,
    messages=[{"role": "user", "content": "Send 'Build passed' to #ci"}],
    tools=tools,
)

for block in response.content:
    if block.type == "tool_use":
        result = fastn.execute(tool=block.name, params=block.input)
        print(result)

Google Gemini Example

from google import genai
from fastn import FastnClient, ToolFormats

fastn = FastnClient()
tools = fastn.get_tools_for("Send a Slack message", format=ToolFormats.gemini, limit=10)

client = genai.Client()
response = client.models.generate_content(
    model="gemini-2.0-flash",
    contents="Send a greeting to #general on Slack",
    # format=ToolFormats.gemini returns function declarations; wrap them in a Tool
    config=genai.types.GenerateContentConfig(
        tools=[genai.types.Tool(function_declarations=tools)],
    ),
)

for part in response.candidates[0].content.parts:
    if fn := part.function_call:
        result = fastn.execute(tool=fn.name, params=dict(fn.args))
        print(result)

AWS Bedrock Example

import boto3
from fastn import FastnClient, ToolFormats

fastn = FastnClient()
tools = fastn.get_tools_for("Notify the team on Slack", format=ToolFormats.bedrock)

bedrock = boto3.client("bedrock-runtime")
response = bedrock.converse(
    modelId="anthropic.claude-sonnet-4-20250514-v1:0",
    messages=[{"role": "user", "content": [{"text": "Notify #ops that deploy is done"}]}],
    toolConfig={"tools": tools},
)

for block in response["output"]["message"]["content"]:
    if "toolUse" in block:
        tool_use = block["toolUse"]
        result = fastn.execute(tool=tool_use["name"], params=tool_use["input"])
        print(result)

Using Connector Names Directly

For local registry lookup without an API call, pass connector names:

from fastn import Connectors, ToolFormats
# Single connector
tools = fastn.get_tools_for("slack tools", connector=Connectors.slack, format=ToolFormats.openai)

# Multiple connectors: limit applies to the total across all
tools = fastn.get_tools_for(
    "project tools",
    connector=[Connectors.slack, Connectors.jira, Connectors.github],
    format=ToolFormats.openai,
    limit=10,
)

Connector Catalog

Inspect the connector registry programmatically:

from fastn import Connectors
# List all connectors in the registry
connectors = fastn.connectors.list()
for c in connectors:
    print(f"{c['name']}: {c['tool_count']} tools")

# Get details for a specific connector
slack = fastn.connectors.get(Connectors.slack)
print(slack["tools"])

# List all tools in a connector with full schemas
tools = fastn.get_tools(Connectors.slack)
for t in tools:
    print(f"  {t['name']}: {t['description']}")

# Get a single tool's schema
tool = fastn.get_tool(Connectors.slack, "send_message")
print(tool["inputSchema"])

Flows

Flows (called workflows in the fastn dashboard) chain tools across connectors. Describe one in plain English and the flow builder agent generates it; then deploy it and run it by name.

from fastn import FlowStatus, Stages
import uuid

# Generate a flow with the flow builder agent. It is a conversation: reuse the
# session id to answer its follow-up questions.
session = str(uuid.uuid4())
reply = fastn.flows.generate(
    "When a Jira ticket is created, post a summary to #engineering on Slack",
    session_id=session,
)
print(reply)  # the agent's answer: the flow it built, or questions for you
fastn.flows.generate("Use the ENG project and the #engineering channel", session_id=session)

# List flows (status filter is case-insensitive: FlowStatus.deployed, FlowStatus.draft, ...)
flows = fastn.flows.list()
deployed = fastn.flows.list(status=FlowStatus.deployed)

# Inspect a flow and the input it expects
flow = fastn.flows.get("jira_to_slack")
schema = fastn.flows.schema("jira_to_slack")   # {"fields": [...], "schema": {...}}

# Deploy, then run by flow name
fastn.flows.deploy("jira_to_slack", stage=Stages.live, comment="first release")
result = fastn.flows.run("jira_to_slack", input_data={"ticket": "ENG-42"}, stage=Stages.live)

# Update the schedule or enable/disable (by flow id), and delete
flow_id = flows[0]["flow_id"]
fastn.flows.update(flow_id, schedule="0 9 * * MON-FRI", enabled=True)
fastn.flows.delete(flow_id)          # also accepts a flow name

Authentication

fastn supports three authentication types at different levels. The SDK triggers connectors and flows; the platform manages OAuth token refresh and credential injection.

1. API Key: App-to-Platform

Static credential that authenticates your application to the fastn platform. Create it in the fastn dashboard; it does not expire.

Your App                          fastn Platform
  │                                     │
  │  POST /executeTool                  │
  │  x-fastn-api-key: sk_live_xxx      │
  │  x-fastn-space-id: proj_abc        │
  │ ──────────────────────────────────→ │
  │                                     │  Validate API key
  │                                     │  Resolve project
  │              200 OK + result        │
  │ ←────────────────────────────────── │

fastn = FastnClient(api_key="sk_live_xxx", project_id="proj_abc")
# or: export FASTN_API_KEY=sk_live_xxx

2. OAuth Token: Developer-to-Platform

JWT obtained via browser-based login (RFC 8628 Device Authorization Grant). Used by the CLI and dashboard. Auto-refreshed with a 30-second expiry buffer.

Developer          Browser            fastn (Keycloak)
  │                   │                      │
  │  fastn login      │                      │
  │ ──────────────────────────────────────→  │
  │                   │    device_code +      │
  │  ←───────────────────  user_code         │
  │                   │                      │
  │  "Visit URL,      │                      │
  │   enter code"     │                      │
  │ ────────────────→ │                      │
  │                   │  User logs in +      │
  │                   │  enters code         │
  │                   │ ──────────────────→  │
  │                   │      authorized      │
  │                   │ ←──────────────────  │
  │                   │                      │
  │  Poll for token   │                      │
  │ ──────────────────────────────────────→  │
  │       access_token + refresh_token       │
  │ ←──────────────────────────────────────  │
  │                                          │
  │  (saved to .fastn/config.json)           │
  │                                          │
  │  POST /executeTool                       │
  │  Authorization: Bearer <jwt>             │
  │ ──────────────────────────────────────→  │
  │                                          │
  │  (on expiry: auto-refresh with 30s       │
  │   buffer using refresh_token)            │

# CLI handles this automatically:
# $ fastn login
# Tokens stored in .fastn/config.json (0o600 permissions)
fastn = FastnClient()  # Picks up auth_token from config

3. Connection: End-User-to-External-Service

OAuth connections that link your end-users to external services (Slack, GitHub, etc.) through the fastn OAuth vault. Tokens are stored server-side, auto-refreshed by the platform, and isolated per tenant. Your code never sees raw credentials.

The @fastn-ai/embed widget captures auth and connector config on the frontend. This SDK triggers connectors and flows on the backend; pass connection_id to route to the right account.

Your App (backend)       Frontend (embed widget)      fastn Platform         External Service
  │                           │                         │                       │
  │  auth.connect(            │                         │                       │
  │    connector="slack"      │                         │                       │
  │    tenant="acme")         │                         │                       │
  │ ───────────────────────────────────────────────→    │                       │
  │                           │                         │                       │
  │  { connection_id,         │                         │                       │
  │    auth_url }             │                         │                       │
  │ ←───────────────────────────────────────────────    │                       │
  │                           │                         │                       │
  │  Pass auth_url to         │                         │                       │
  │  frontend ──────────────→ │                         │                       │
  │                           │  Render login /         │                       │
  │                           │  redirect user          │                       │
  │                           │ ────────────────────────────────────────────→   │
  │                           │                         │    User authorizes    │
  │                           │                         │ ←─────────────────    │
  │                           │  Callback received      │  Store tokens in      │
  │                           │ ←───────────────────    │  OAuth vault          │
  │                           │                         │                       │
  │  auth.status(             │                         │                       │
  │    connection_id)         │                         │                       │
  │ ───────────────────────────────────────────────→    │                       │
  │  { status: "authorized" } │                         │                       │
  │ ←───────────────────────────────────────────────    │                       │
  │                           │                         │                       │
  │  execute(tool,            │                         │                       │
  │    connection_id)         │                         │                       │
  │ ───────────────────────────────────────────────→    │                       │
  │                           │                         │  Inject credentials   │
  │                           │                         │ ──────────────────→   │
  │                           │                         │       result          │
  │                           │                         │ ←──────────────────   │
  │       result              │                         │                       │
  │ ←───────────────────────────────────────────────    │                       │
  │                           │                         │                       │
  │  (on token expiry: platform auto-refreshes)         │                       │

from fastn import Connectors
# 1. Start an OAuth connection for your end-user
result = fastn.auth.connect(
    connector=Connectors.slack,
    tenant_id="customer_acme",
    redirect_url="https://myapp.com/callback",
)
# Send the user to result["auth_url"] (or let the embed widget handle it)

# 2. Check connection status (after user completes auth)
status = fastn.auth.status(connection_id=result["connection_id"])
print(status["status"])  # "pending", "authorized", "expired"

# Or look up by connector + tenant
status = fastn.auth.status(connector=Connectors.github, tenant_id="customer_acme")

# 3. Use the connection: the platform injects the credentials
fastn.slack.send_message(
    connection_id=result["connection_id"],
    channel="general", text="Hello!",
)

Custom Auth Provider

For apps with their own identity provider, configure fastn to validate end-user tokens via your userinfo endpoint.

fastn.auth.configure_custom(
    userinfo_url="https://myapp.auth0.com/userinfo",
)

Multi-Connection Support

When you have multiple connections for the same connector (e.g. two Slack workspaces):

# Per-call
fastn.slack.send_message(connection_id="conn_workspace_a", channel="general", text="Hi!")

# Or bind once
slack_a = fastn.connect("conn_workspace_a")
slack_a.send_message(channel="general", text="Hello!")

Multi-Tenant Support

Route requests to the correct tenant (customer, organization, or team):

# Per-call override
fastn.slack.send_message(tenant_id="acme", channel="general", text="Hello!")

# Constructor-level default
fastn = FastnClient(tenant_id="acme")

# Environment variable
# export FASTN_TENANT_ID=acme

# CLI flag
# fastn connector run slack send_message --tenant acme --channel general --text "Hello!"

Priority: per-call tenant_id > CLI --tenant flag > constructor param > FASTN_TENANT_ID env var > config file

Environments

Switch between LIVE, STAGING, and DEV:

from fastn import Stages
dev = FastnClient(stage=Stages.dev)
staging = FastnClient(stage=Stages.staging)
prod = FastnClient()  # defaults to LIVE

Or: export FASTN_STAGE=DEV

AI-Powered Mode

For quick prototyping, use natural language:

result = fastn.run("Send hello to #general on Slack")

Projects

List available projects for the authenticated user:

projects = fastn.projects.list()
for p in projects:
    print(f"{p['name']} ({p['id']})")

CLI Reference

fastn agent

Agentic CLI: describe what you want in natural language and the agent discovers tools, sends them to your LLM via native function calling, and executes tool calls in a loop until the task is complete.

fastn agent "Send hello to #general on Slack"
fastn agent --connector slack "List all channels"
fastn agent --eval "Create a Jira ticket for the login bug"

The first run asks you to pick an LLM provider and enter its API key (saved to .fastn/config.json).

fastn connector run

Execute connector tools directly from the command line:

# List available tools
fastn connector run slack

# Interactive mode (prompts for each parameter)
fastn connector run slack send_message

# Inline parameters
fastn connector run slack send_message --channel general --text "Hello!"

# With tenant override (or pass the tenant as a third positional argument)
fastn connector run slack send_message --tenant acme --channel general --text "Hello!"

API Reference

FastnClient

FastnClient(
    api_key: str = None,        # fastn API key (or FASTN_API_KEY)
    project_id: str = None,     # Project ID (or FASTN_PROJECT_ID)
    auth_token: str = None,     # JWT from fastn login (or FASTN_AUTH_TOKEN)
    tenant_id: str = None,      # Tenant ID (or FASTN_TENANT_ID)
    stage: str = None,          # "LIVE", "STAGING", "DEV" (or FASTN_STAGE)
    agent_id: str = None,       # Agent ID, for agent-scoped calls
    config_path: str = None,    # Path to config.json
    timeout: float = None,      # HTTP timeout in seconds (default from config, else 30)
    max_retries: int = 3,       # Retry count for transient failures
    verbose: bool = False,      # Debug logging
)

Data Plane (tool execution):

Control Plane (discovery):

Flows:

Auth:

Skills:

Projects:

Connect Kit:

Requests are retried on transient failures (max_retries, default 3, with exponential backoff).

AsyncFastnClient

Same API as FastnClient, but all methods are async/await:

async with AsyncFastnClient() as fastn:
    result = await fastn.slack.send_message(channel="general", text="Hi")
    flows = await fastn.flows.list()
    skills = await fastn.skills.list()
    status = await fastn.auth.status(connection_id="conn_abc")

Error Handling

All exceptions inherit from FastnError:

from fastn import (
    FastnClient, FastnError, AuthError, ConfigError, APIError,
    ConnectorNotFoundError, ToolNotFoundError, ConnectionNotFoundError,
    OAuthError, RegistryError, FlowNotFoundError, RunNotFoundError,
)

try:
    fastn.slack.send_message(channel="general", text="Hello!")
except AuthError:
    print("Invalid credentials: check your API key")
except ConnectorNotFoundError as e:
    print(f"Connector '{e.connector_name}' not found: run fastn connector add {e.connector_name}")
except ToolNotFoundError as e:
    print(f"Tool '{e.tool_name}' not found in '{e.connector_name}'")
except APIError as e:
    print(f"HTTP {e.status_code}: {e}")
except ConfigError:
    print("Run: fastn login")

IDE Autocomplete

Type stubs for the client and 250+ connectors ship in the package, so autocomplete works after pip install fastn-ai:

• PyCharm / IntelliJ: Works automatically
• VS Code (Pylance): Works automatically
• mypy: Works automatically (PEP 561 py.typed marker included)

To refresh stubs with the latest connector schemas, run fastn connector sync.

Examples

Runnable scripts are in the repository:

Platform API client (fastn.platform)

The fastn platform REST API for Python: every endpoint, generated from fastn's OpenAPI spec, plus a small hand-written layer that makes the common partner flow easy. It mirrors the platform client in the Node SDK (@fastn-ai/sdk): every operation is a snake_case method on its area (fastn.embed.create_setup_link) with the operationId as an alias (createSetupLink); per-call options are keyword arguments. The Node SDK's workspace tool calls (ToolsClient, fastn-sdk-tools) are Node-only.

This is a different FastnClient from fastn.FastnClient above: it takes a platform API key from the fastn dashboard and talks to https://api.fastn.dev. It is synchronous.

Apps, Categories, Entities, HubSpotEvents, EventTypes and UnifiedActions name every app in the unified catalog, every category and entity, HubSpot's change events, and the events your webhook receives, so calls are completed by your editor instead of typed as strings: customer.connect(Apps.googleCalendar), customer.watch(Apps.hubspot, [HubSpotEvents.contactCreated]), fastn.catalog.fields(Categories.crm, Entities.contact), provider=Apps.hubspot. Each value is the exact string the API takes, so existing string calls keep working. The lists are the unified catalog on production as of 0.5.0; an app, category or entity outside them can still be passed as a string. HubSpotEvents.dealStageChanged delivers every deal property change: keep stage moves with propertyName == "dealstage". The snippets below import them from fastn.platform.

import os
from fastn.platform import Apps, Categories, Entities, EventTypes, FastnClient, HubSpotEvents

fastn = FastnClient(
    api_key=os.environ["FASTN_API_KEY"],
    # base_url="https://api.fastn.dev", timeout=60.0, retries=2,
    # on_request=lambda log: print(log), defaults={...}
)

# 1. Register a customer under your account, using YOUR id for them.
customer = fastn.customer("acme-42")
customer.register({"name": "Acme Dental"})                        # idempotent: returns the existing org on repeat

# 2. Give them a link to connect their tools (Google Calendar, HubSpot, ...), by connector slug.
link = customer.connect(Apps.googleCalendar)
print(link["url"])                                                # send this to the customer

# 3. Typed methods on whatever the customer connected: the provider is picked from their connections.
booked = customer.calendar.book(
    {"title": "Intro call", "start_time": "2026-11-03T15:00:00Z", "end_time": "2026-11-03T15:30:00Z", "timezone": "Europe/London", "attendee_email": "jane@example.com"},
    idempotency_key="booking-123", correlation_id="req-abc",      # safe to retry; traceable later
)
print(booked["id"], booked["provider"], booked["data"].get("web_url"))

customer.calendar.reschedule(booked["id"], {"start_time": "2026-11-03T16:00:00Z", "end_time": "2026-11-03T16:30:00Z"})
customer.calendar.cancel(booked["id"], idempotency_key="cancel-123")

page = customer.crm.contacts.find({"email": "jane@example.com"})  # one page (the search route has no cursor); list() pages
contact = page["records"][0] if page["records"] else customer.crm.contacts.create({"first_name": "Jane", "last_name": "Doe", "email": "jane@example.com"})
customer.crm.notes.add({"body": "Booked an intro call", "parent_id": contact["id"], "parent_type": Entities.contact})

# 4. What did fastn do for my request id?
customer.call_log("req-abc")

# Changes at an app, sent to your webhook as connector.event / unified.event: watch them per customer.
account_id = "..."                                                # one of the customer's accounts, from customer.connections()
w = customer.watch(Apps.hubspot, [HubSpotEvents.contactCreated, HubSpotEvents.dealStageChanged], account_id=account_id)
customer.watches(Apps.hubspot)                                       # [{"id", "app", "events", "status", ...}]
customer.unwatch(Apps.hubspot)                                       # or unwatch(w["id"])

customer.disconnect(account_id)                                   # the customer gets connection.revoked
customer.rename_account(account_id, "Front desk calendar")
customer.update({"name": "Acme Dental Ltd"})                      # or {"status": "suspended"}

# Your webhook, once for the whole account.
hook = fastn.webhooks.create({"url": "https://app.acme.com/fastn", "eventTypes": [EventTypes.connectionConnected, EventTypes.setupLinkCompleted, EventTypes.connectorEvent, EventTypes.unifiedEvent]})
hook["secret"]                                                    # shown once: the signing secret for parse_event
fastn.webhooks.test(hook["id"]); fastn.webhooks.deliveries(hook["id"]); fastn.webhooks.rotate_secret(hook["id"])

Calendar and CRM, typed

customer.calendar and customer.crm are the unified API with the fields named (the catalog's own names, start_time, first_name, ...; datetime values are sent as ISO strings). Answers are {"id", "provider", "data", "warnings"?} records or {"records", "nextCursor", "hasMore"} pages. The provider is the one the customer has connected for that entity: no connected provider raises 404 NO_CONNECTED_PROVIDER (send them a setup link), several raise 409 PROVIDER_AMBIGUOUS (pass provider=). warnings lists fields the provider does not map and therefore dropped. Every method also takes provider=, connection_id=, idempotency_key=, correlation_id=, ...

find(filters, page_size=) answers one page (the search route has no cursor); list(page_size=, cursor=) pages. What the typed methods do not cover is still one call away: customer.records(Categories.crm, Entities.contact, provider=Apps.hubspot) takes any entity in the catalog with untyped input (create, get, list, update, delete, plus search(filters) and providers()), and customer.run_action(connector, action, input) runs a connector's own action.

# The catalog: which categories and entities exist, the fields each takes, which providers a customer connected.
categories = fastn.catalog.categories()                 # [{"slug": "crm", "entities": [...]}]
fields = fastn.catalog.fields(Categories.crm, Entities.contact)         # [{"field": "email", "type": "string", "required": False}, ...]
providers = customer.catalog.providers(Categories.crm, Entities.contact)  # [{"connectorSlug": "hubspot", "connected": True, ...}]

# Anything the unified API does not cover: the connector's own action with the customer's connection.
r = customer.run_action(Apps.hubspot, "updateDeal", {"dealId": deal_id, "properties": {"dealstage": "qualifiedtobuy"}})

Every area is on the client directly, with typed inputs. The partner flow lives in orgs, embed, connections, unified, events and workflows; platform-administration and backend-only routes are not generated, because a partner key cannot call them.

fastn.orgs.list_orgs({"type": OrgTypes.endOrg})
fastn.connections.list_connections({"endOrgId": customer.id()})
fastn.embed.get_setup_link({"id": "stp_..."})
fastn.orgs.get_tenant({"id": "..."})                    # every operation is a typed method on its area
fastn.me()                                              # who this key is: actor, role, org, parentOrg

Customer ids, explained once

Use fastn.customer(ref) (or the customer= option) and nothing else. The raw API names the same customer differently per operation; those generated input fields still exist, but they are the raw API. Scoping calls to a customer needs a multi-tenant partner API key.

customer.find() returns the customer's org row (or None when fastn does not know the ref). register, connect, links, watch, disconnect, rename_account, update, call_log, setup_link, embed_token, connect_with_credentials, connections, get, remove, calendar, crm and records all take the ref and resolve fastn's id only where the API insists on it. The lookup is limited to orgs under your own org, so a key that can see more than your account can still never resolve someone else's customer.

customer.api.<area>.<operation>() runs any operation for that customer: it sends x-end-org-ref and, on routes that take the customer in the body instead (OAuth initiation, stored connections, event subscriptions, secrets, setup links, ...), fills that field in too.

Connecting a customer's apps

# A setup link by app: the account's widget for that connector is found for you (several: pass widget_id=).
link = customer.connect(Apps.hubspot, connection_name="Main CRM", redirect_url="https://app.acme.com/connected")
link["url"]                                             # then customer.links.get(link["setupId"]) / links.cancel(...)

# Or start OAuth directly, without a widget:
answer = customer.connect({"connectorId": Apps.googleCalendar, "connectionName": "Google Calendar", "returnUrl": "https://app.acme.com/connected"})

# Typed-in credentials (API key, basic auth): the fields come from the connector's auth method. The platform checks
# them; an incomplete set raises 400 CREDENTIALS_INCOMPLETE with the missing names (err.body["missing"]).
fields = customer.auth_fields("acme-crm")              # {"names", "required", "schema", "authMethodId", ...}
customer.connect_with_credentials({"connectorId": "acme-crm", "credentials": {"apiKey": "..."}, "name": "Acme CRM"})

Prefer setup links for credentials: fastn's hosted page collects them, so they never pass through your backend.

Watching changes

customer.watch(app, events, account_id=, connection_id=, name=) turns on an app's change events for a customer; each change reaches your webhook as connector.event and unified.event. Name events in plain words (EVENT_ALIASES: contact.created, deal.stageChanged, ...) or as the app names them; an unknown name raises 400 EVENT_NOT_FOUND with the valid list (customer.watchable_events(app)). No account on the app raises 404 CONNECTION_NOT_FOUND, several raise 409 CONNECTION_AMBIGUOUS (pass connection_id=; one of another app is 400 CONNECTION_MISMATCH). watches(app=None) lists every page, and unwatch(app | id) removes (an id that is not this customer's is 404 WATCH_NOT_FOUND). disconnect and rename_account refuse an account id that is not the customer's. HubSpot has no stage-only event: deal.stageChanged is deal.propertyChange, so check propertyName == "dealstage" in the event.

Responses and errors

Answers are returned as the REST API sends them (usually {"data": ...}; unified records are {"id", "provider", "data"}), so the SDK agrees 1:1 with the REST docs, typed as TypedDicts under fastn.platform.types.

Non-2xx answers raise FastnApiError with status, code (e.g. NOT_FOUND, VALIDATION_ERROR, NO_CONNECTED_PROVIDER, OPERATION_NOT_SUPPORTED, PROVIDER_AMBIGUOUS), message, details, body and request_id (quote it when asking fastn for help), plus is_not_found, is_conflict, is_rate_limited, is_auth and is_validation. Network failures raise FastnNetworkError. GET calls are retried on 429/502/503/504 (default 2 retries, honouring Retry-After). A write is retried only when it carries an idempotency_key and the platform de-duplicates that operation by it (unified record writes, workflow execute); every other write is never retried, so a lost answer cannot make it act twice.

from fastn.platform import FastnApiError

try:
    fastn.orgs.get_tenant({"id": tenant_id})
except FastnApiError as e:
    if e.is_not_found:
        ...

Per-call options

Every method takes keyword options: idempotency_key, correlation_id, customer={"ref": ...} or {"id": ...}, installation_id, target_org, headers, extra_body, extra_query, timeout, retries. extra_body and extra_query send fields the SDK does not know yet; they are passed as given. FastnClient(defaults={...}) applies options to every call, e.g. defaults={"customer": {"ref": "acme-42"}} to scope a whole client to one customer.

Events (webhooks)

from fastn.platform import Apps, EventTypes, parse_event

# Flask: event = parse_event(os.environ["FASTN_EVENT_SECRET"], request.headers, request.get_data())
event = parse_event(secret, headers, raw_body)          # raises ValueError on a bad signature
# event["id"] is stable across retries: drop repeats by id.
if event["type"] == EventTypes.connectionConnected and event.get("app") == Apps.hubspot:
    ...                                                 # start a watch
elif event["type"] == EventTypes.setupLinkCompleted:
    ...                                                 # event["connections"]: app -> account id

The signature is sha256=HMAC-SHA256(secret, "<X-Fastn-Timestamp>.<raw body>"); deliveries older than 5 minutes are rejected (tolerance_seconds=, 0 disables the check). Always pass the raw body, never a re-serialised object. verify_event_signature answers a bool instead, read_event_headers reads the fastn headers, sign_event_body signs a body the same way (for tests), and internal_ids=True keeps fastn's internal customer id in a parsed event.

Pagination and helpers

from fastn.platform import EventOutcomes, paginate

for ev in paginate(fastn.events.list_events, {"outcome": EventOutcomes.sentToDlq}, page_size=100):
    ...

Works for limit/offset lists and cursor lists. Also exported: free_intervals and slots_of (the slot maths behind calendar.free_slots).

FAQ

What is fastn? The embedded integration platform for B2B SaaS. Your customers connect their own apps (HubSpot, Google Calendar, Slack, and 1,000+ more) inside your product, and your code and AI agents use those connections through one API.

How do I add customer-facing integrations to my SaaS app in Python? Install fastn-ai, register each customer with fastn.customer(your_id).register(...), let them connect apps with a setup link (customer.connect(app)) or the @fastn-ai/embed widget, then call customer.crm, customer.calendar, or customer.run_action(...). See Embed fastn in your product.

Does fastn support multi-tenant apps? Yes. Every call is scoped to one customer, addressed by your own id; credentials are stored and refreshed per customer; a customer can connect several accounts of one app.

Can AI agents like Claude use fastn integrations? Yes: as function-calling tools from this SDK (get_tools_for, execute) for OpenAI, Anthropic, Gemini, and Bedrock, or through the fastn MCP gateway from Claude, Cursor, or any MCP client.

Is there a Node.js / TypeScript SDK? Yes: @fastn-ai/sdk, with the same platform API client as fastn.platform.

Development

Working on the SDK itself (from a clone of the repository):

# Install in dev mode
pip install -e ".[dev]"

# Run all tests
make test

# Run only SDK tests
make test-sdk

# Run only CLI tests
make test-cli

# Run a single test file
make test-file F=tests/cli/test_cli_commands.py

# Or use pytest directly
python3 -m pytest tests/sdk/ -q       # SDK only
python3 -m pytest tests/cli/ -q       # CLI only
python3 -m pytest tests/ -q           # All

Generated code is checked in CI: python3 scripts/generate_platform.py --check (platform client) and python3 scripts/generate_client_stub.py --check (the fastn/__init__.pyi type stub). The platform live suite runs with LIVE=1 FASTN_API_KEY=... FASTN_TEST_CUSTOMER=... pytest tests/platform/live against api.fastn-test.dev.

Test Structure

tests/
├── sdk/                           # SDK core tests
│   ├── test_client.py             # FastnClient, AsyncFastnClient
│   ├── test_connector.py          # DynamicConnector proxy
│   ├── test_config.py             # Config loading, env vars, validation
│   ├── test_exceptions.py         # Exception hierarchy
│   ├── test_auth.py               # Auth helpers
│   ├── test_oauth.py              # OAuth device flow, token refresh
│   ├── test_projects.py           # Projects namespace
│   └── test_skills.py             # Skills namespace
├── platform/                      # fastn.platform tests (live/ runs against the API)
└── cli/                           # CLI command tests
    ├── test_cli_commands.py        # All CLI commands (sync, add, remove, list, run, agent, etc.)
    ├── test_cli_helpers.py         # CLI helper functions
    ├── test_agent_command.py       # fastn agent command
    ├── test_agent_helpers.py       # Agent helper functions
    ├── test_helpers_extended.py    # Extended helper tests (token, schema, parsing)
    ├── test_skills_command.py      # fastn skill command
    ├── test_detect_api_error.py    # API error detection across providers
    └── test_detect_languages.py    # SDK language detection for stub generation

License

MIT