Python SDK

truto-python-sdk is an async Python client for the Truto API. It covers the Unified, Proxy, and Custom APIs, MCP, and the Admin API endpoints in the Truto API Reference.

Package: pypi.org/project/truto-python-sdk. The methods on this page require version 0.2.0 or later.

Installation

pip install "truto-python-sdk>=0.2.0"

Quick start

Create an API token, then:

import asyncio
from truto_python_sdk import TrutoApi
 
truto_api = TrutoApi(token="<your_api_token>")
 
async def main():
    # Installed integrations in the environment
    async for integration in truto_api.environment_integrations.list():
        print(integration)
 
    # Integrated accounts for a tenant
    accounts = await truto_api.integrated_accounts.list(tenant_id="acme-1").to_array()
 
    # A Unified API call
    async for contact in truto_api.unified_api.list(
        "crm", "contacts", {"integrated_account_id": accounts[0]["id"]}
    ):
        print(contact)
 
asyncio.run(main())

How the client works

Async. Every method returns a coroutine you await, except list(), which returns a paginated cursor (see Pagination).
Resources. Each Admin API resource is an attribute of TrutoApi, such as truto_api.tenants or truto_api.sync_jobs. Most expose list, get, create, update, and delete.
Request bodies are plain dicts, matching the JSON bodies in the API reference.
Query parameters are keyword arguments on list() (truto_api.tenants.list(environment_id="...")) and on the Admin API methods whose signature includes **query_params; many get, create, update and delete methods take none. The Unified and Proxy APIs take a query_params dict.
Errors. A non-2xx response raises aiohttp.ClientResponseError, with the HTTP status on .status.
Base URL. It defaults to https://api.truto.one. Pass base_url to TrutoApi to use a different one.

import aiohttp
 
try:
    await truto_api.integrated_accounts.get("missing-id")
except aiohttp.ClientResponseError as error:
    print(error.status)  # 404

The Python SDK sends each request once. It does not retry, and it does not wait on 429 or Retry-After — a rate-limited call raises straight away. Keep that in mind against the rate limits: 100 requests per second per API token, and 50 requests per 10 seconds per integrated account.

No retries. Any non-2xx response, including 429, 502, 503, and 504, raises aiohttp.ClientResponseError on the first attempt.
What the error carries. .status (the HTTP status), .message (the reason phrase, such as Too Many Requests), and .headers (the response headers, so Retry-After is readable). The response body is not on the error, so Truto's message and truto_error_insight fields aren't available from it.
Timeout. Each request has aiohttp's default timeout: 300 seconds in total, 30 seconds to connect. The SDK has no option to change it; wrap the call in asyncio.timeout() to set a shorter limit.
One session per call. Every request opens and closes its own HTTP session, so connections aren't reused between calls.

This wrapper retries the statuses the retry strategy marks as safe, waits for Retry-After when the response has one, and backs off exponentially when it doesn't. Because the error carries no body, it can't read truto_error_code, so it also retries a 502 with upstream_response_too_large, which fails the same way every time. Pass it a function that starts the call, so each attempt sends a fresh request:

import asyncio
import random
 
import aiohttp
from truto_python_sdk import TrutoApi
 
RETRY_STATUSES = {429, 502, 503, 504}
 
 
async def with_retry(make_call, max_retries=3, default_wait=10):
    for attempt in range(max_retries + 1):
        try:
            return await make_call()
        except aiohttp.ClientResponseError as error:
            if error.status not in RETRY_STATUSES or attempt == max_retries:
                raise
            retry_after = (error.headers or {}).get("Retry-After", "")
            if retry_after.isdigit():
                wait = int(retry_after)
            elif error.status == 429:
                wait = default_wait
            else:
                wait = min(2 ** attempt, 30) + random.random()
            await asyncio.sleep(wait)
 
 
truto_api = TrutoApi(token="<your_api_token>")
 
 
async def main():
    tenant = await with_retry(lambda: truto_api.tenants.get("acme-1"))
 
    # Retry page by page; a failed page is fetched again from the same cursor
    cursor = truto_api.tenants.list(limit=50)
    while True:
        page = await with_retry(cursor.next_page)
        print(page)
        if cursor.next_cursor is None:
            break
 
    # A 30-second limit for one call, retries included
    async with asyncio.timeout(30):
        await with_retry(lambda: truto_api.integrated_accounts.get("ia-id"))
 
 
asyncio.run(main())

Wrap reads freely. Wrap a create or other write only when sending it twice is harmless: a 502 or 504 can arrive after the provider has already applied the change.

Pagination

list() returns a cursor that follows next_cursor for you. Iterate it with async for:

async for ticket in truto_api.unified_api.list(
    "ticketing", "tickets", {"integrated_account_id": "766cc1ee-..."}
):
    print(ticket)

Collect every page with to_array():

tickets = await truto_api.unified_api.list(
    "ticketing", "tickets", {"integrated_account_id": "766cc1ee-..."}
).to_array()

Or fetch one page at a time with next_page(). Stop when the cursor's next_cursor is None; calling next_page() after the last page returns the last page again.

cursor = truto_api.tenants.list(limit=50)
while True:
    page = await cursor.next_page()
    print(page)
    if cursor.next_cursor is None:
        break

Unified API

Accessible via truto_api.unified_api. Pass integrated_account_id in query_params. See the Unified API reference for the models and resources available.

query = {"integrated_account_id": "766cc1ee-..."}
 
contacts = truto_api.unified_api.list("crm", "contacts", query)
contact = await truto_api.unified_api.get("crm", "contacts", "123", query)
created = await truto_api.unified_api.create("crm", "contacts", {"first_name": "Ada"}, query)
await truto_api.unified_api.update("crm", "contacts", "123", {"first_name": "Grace"}, query)
await truto_api.unified_api.delete("crm", "contacts", "123", query)
 
# Custom methods: POST /unified/{model}/{resource}/{method}
results = await truto_api.unified_api.custom_method("crm", "contacts", "search", {"query": "acme"}, query)

Method	Endpoint
`list(model, resource, query_params=None)`	`GET /unified/{model}/{resource}`
`get(model, resource, id, query_params=None)`	`GET /unified/{model}/{resource}/{id}`
`create(model, resource, data, query_params=None)`	`POST /unified/{model}/{resource}`
`update(model, resource, id, data, query_params=None)`	`PATCH /unified/{model}/{resource}/{id}`
`delete(model, resource, id, query_params=None)`	`DELETE /unified/{model}/{resource}/{id}`
`custom_method(model, resource, method, data, query_params=None)`	`POST /unified/{model}/{resource}/{method}`
`meta(model, resource, integration_name, method)`	`GET /unified/{model}/{resource}/{integration_name}/meta/{method}`
`model_meta(model, integration_name)`	`GET /unified/meta/{model}/{integration_name}`

meta returns an integration's mapping metadata (schema and supported parameters) for one unified method. model_meta returns the resources and methods of a unified model that an integration supports.

meta = await truto_api.unified_api.meta("crm", "contacts", "hubspot", "list")
crm_on_hubspot = await truto_api.unified_api.model_meta("crm", "hubspot")

Proxy API

Accessible via truto_api.proxy_api. Calls go to the integration's own API through Truto, with resource names as defined on the integration.

query = {"integrated_account_id": "766cc1ee-..."}
 
tickets = truto_api.proxy_api.list("tickets", query)
ticket = await truto_api.proxy_api.get("tickets", "123", query)
await truto_api.proxy_api.create("tickets", {"subject": "Help"}, query)
await truto_api.proxy_api.update("tickets", "123", {"status": "closed"}, query)
await truto_api.proxy_api.delete("tickets", "123", query)
await truto_api.proxy_api.custom_method("tickets", "merge", {"ids": ["1", "2"]}, query)

Method	Endpoint
`list(resource, query_params=None)`	`GET /proxy/{resource}`
`get(resource, id, query_params=None)`	`GET /proxy/{resource}/{id}`
`create(resource, data, query_params=None)`	`POST /proxy/{resource}`
`update(resource, id, data, query_params=None)`	`PATCH /proxy/{resource}/{id}`
`delete(resource, id, query_params=None)`	`DELETE /proxy/{resource}/{id}`
`custom_method(resource, method, data, query_params=None)`	`POST /proxy/{resource}/{method}`

Custom API

Accessible via truto_api.custom_api. Sends an arbitrary request to the integration's API through Truto, at /custom/{path}. Query parameters are keyword arguments.

result = await truto_api.custom_api.get("my/custom/path", integrated_account_id="766cc1ee-...")
await truto_api.custom_api.post("my/custom/path", {"key": "value"}, integrated_account_id="766cc1ee-...")

Method	Endpoint
`get(path, **query_params)`	`GET /custom/{path}`
`post(path, data=None, **query_params)`	`POST /custom/{path}`
`put(path, data=None, **query_params)`	`PUT /custom/{path}`
`patch(path, data=None, **query_params)`	`PATCH /custom/{path}`
`delete(path, **query_params)`	`DELETE /custom/{path}`

MCP

Accessible via truto_api.mcp. Talks to an integrated account's MCP server, using the token from Create MCP server.

mcp_token = "your-mcp-token"
 
await truto_api.mcp.initialize(mcp_token, "my-client", "1.0.0")
tools = await truto_api.mcp.list_tools(mcp_token)
result = await truto_api.mcp.call_tool(mcp_token, "list_tickets", {"status": "open"})
 
# Any other JSON-RPC message
await truto_api.mcp.call(mcp_token, {"method": "tools/list", "id": 3})

Method	Description
`initialize(token, client_name, client_version)`	Sends `initialize`
`list_tools(token)`	Sends `tools/list`
`call_tool(token, name, args)`	Sends `tools/call`
`call(token, payload)`	Sends a raw JSON-RPC payload to `POST /mcp/{token}`

Integrated accounts

Accessible via truto_api.integrated_accounts. API reference.

accounts = truto_api.integrated_accounts.list(tenant_id="acme-1")
account = await truto_api.integrated_accounts.get("ia-id")
 
await truto_api.integrated_accounts.update_credentials("ia-id", {"context": {"api_key": "new-key"}})
await truto_api.integrated_accounts.validate("ia-id")
await truto_api.integrated_accounts.bulk_delete({"tenant_id": "acme-1"})
capabilities = await truto_api.integrated_accounts.capabilities("ia-id")

Method	Endpoint
`list(**query_params)`	`GET /integrated-account`
`get(id)`	`GET /integrated-account/{id}`
`create(data)`	`POST /integrated-account` (import an account with existing credentials)
`update(id, data)`	`PATCH /integrated-account/{id}`
`delete(id)`	`DELETE /integrated-account/{id}`
`update_credentials(id, data)`	`PATCH /integrated-account/{id}/credentials` (send the new credentials under `context`)
`bulk_delete(data)`	`POST /integrated-account/bulk-delete` (by `ids` or `tenant_id`)
`create_integrated_account_token(id)`	`POST /integrated-account/token`
`refresh_credentials(id)`	`POST /integrated-account/refresh-credentials`
`refresh_mapping_cache(data=None)`	`POST /integrated-account/refresh-mapping-cache`
`create_form_token(id)`	`POST /integrated-account/form-token` (tenant session tokens only; an API token gets `403`)
`run_post_install_actions(id)`	`POST /integrated-account/run-post-install-actions`
`validate(id)`	`POST /integrated-account/{id}/validate`
`capabilities(id)`	`GET /integrated-account/{id}/capabilities`
`tools(id)`	`GET /integrated-account/{id}/tools` (deprecated; use `capabilities`)

MCP servers

Each integrated account can have MCP servers. API reference.

server = await truto_api.integrated_accounts.create_mcp_server(
    "ia-id", {"name": "Support agent", "config": {"methods": ["list", "get"]}}
)
servers = await truto_api.integrated_accounts.list_mcp_servers("ia-id")

Method	Endpoint
`list_mcp_servers(ia_id)`	`GET /integrated-account/{ia_id}/mcp`
`create_mcp_server(ia_id, data)`	`POST /integrated-account/{ia_id}/mcp`
`get_mcp_server(ia_id, id)`	`GET /integrated-account/{ia_id}/mcp/{id}`
`update_mcp_server(ia_id, id, data)`	`PATCH /integrated-account/{ia_id}/mcp/{id}`
`delete_mcp_server(ia_id, id)`	`DELETE /integrated-account/{ia_id}/mcp/{id}`

Link tokens

Accessible via truto_api.link_tokens. API reference.

link_token = await truto_api.link_tokens.create({"tenant_id": "acme-1"})

Method	Endpoint
`create(data)`	`POST /link-token`

Tenants

Accessible via truto_api.tenants. API reference.

await truto_api.tenants.create({"id": "acme-1", "name": "Acme"})
await truto_api.tenants.bulk_create({"mode": "upsert", "tenants": [{"id": "acme-2", "name": "Acme 2"}]})

Method	Endpoint
`list`, `get`, `create`, `update`, `delete`	`/tenant`, `/tenant/{id}`
`bulk_create(data)`	`POST /tenant/bulk`

Integrations

Accessible via truto_api.integrations. API reference.

integration = await truto_api.integrations.get("integration-id")
capabilities = await truto_api.integrations.capabilities("hubspot")  # ID or name
 
await truto_api.integrations.install("integration-id")
await truto_api.integrations.uninstall("integration-id")

Method	Endpoint
`list`, `get`, `create`, `update`, `delete`	`/integration`, `/integration/{id}`
`capabilities(id_or_name)`	`GET /integration/{id}/capabilities`
`tools(id_or_name)`	`GET /integration/{id}/tools` (deprecated; use `capabilities`)
`unified_apis(id_or_name)`	`GET /integration/{id}/unified-apis` (deprecated; use `capabilities`)
`install(integration_id, is_enabled=True)`	`POST /environment-integration`
`uninstall(integration_id)`	Finds the integration's environment integration and deletes it

Environment integrations

Integrations installed in your environment. Accessible via truto_api.environment_integrations. API reference.

Method	Endpoint
`list(**query_params)`	`GET /environment-integration`
`get(id)`	`GET /environment-integration/{id}`
`create(data)`	`POST /environment-integration`
`update(id, data)`	`PATCH /environment-integration/{id}`
`delete(id)`	`DELETE /environment-integration/{id}`

Unified models

Accessible via truto_api.unified_models. API reference.

Method	Endpoint
`list`, `get`, `create`, `update`, `delete`	`/unified-model`, `/unified-model/{id}`
`install(unified_model_id)`	`POST /environment-unified-model`
`uninstall(unified_model_id)`	Finds the model's environment unified model and deletes it

Environment unified models

Unified models installed in your environment. Accessible via truto_api.environment_unified_models. API reference.

Method	Endpoint
`list`, `get`, `create`, `update`, `delete`	`/environment-unified-model`, `/environment-unified-model/{id}`

Unified model mappings

How a unified model's resource methods map to an integration. Both resources expose list, get, create, update, and delete.

Attribute	Endpoint	Reference
`truto_api.unified_model_resource_methods`	`/unified-model-resource-method`	Unified model mappings
`truto_api.environment_unified_model_resource_methods`	`/environment-unified-model-resource-method`	Environment unified model mappings

overrides = truto_api.environment_unified_model_resource_methods.list(
    environment_unified_model_id="..."
)
async for override in overrides:
    print(override)

Sync jobs

Each resource below exposes list, get, create, update, and delete.

Attribute	Endpoint	Extra methods	Reference
`truto_api.sync_jobs`	`/sync-job`		Sync jobs
`truto_api.sync_job_runs`	`/sync-job-run`		Sync job runs
`truto_api.sync_job_cron_triggers`	`/sync-job-cron-trigger`	`schedule(id)`: `POST /sync-job-cron-trigger/{id}/schedule`	Sync job cron triggers
`truto_api.sync_job_templates`	`/sync-job-template`		Sync job templates
`truto_api.sync_job_run_states`	`/sync-job-run-state`		Sync job run state

run = await truto_api.sync_job_runs.create({
    "sync_job_id": "sync-job-id",
    "integrated_account_id": "766cc1ee-...",
})
await truto_api.sync_job_runs.delete(run["id"])  # stops the run

Webhooks

Accessible via truto_api.webhooks. API reference.

webhook = await truto_api.webhooks.create({"target_url": "https://example.com/truto"})
await truto_api.webhooks.test(webhook["id"], event_type="integrated_account:created")

Method	Endpoint
`list`, `get`, `create`, `update`, `delete`	`/webhook`, `/webhook/{id}`
`test(id, event_type=None, integrated_account_id=None)`	`POST /webhook/test`

Workflows

Accessible via truto_api.workflows and truto_api.workflow_runs. API reference: Workflows, Workflow runs.

workflow = await truto_api.workflows.create({
    "trigger_name": "integrated_account:active",
    "config": {
        "steps": [
            {
                "type": "run",
                "action": "run_sync_job",
                "config": '{"sync_job_id": "sync-job-id", "integrated_account_id": integrated_account_id}',
            }
        ]
    },
})
async for run in truto_api.workflow_runs.list(workflow_id=workflow["id"]):
    print(run.get("status"))

Attribute	Methods	Endpoint
`truto_api.workflows`	`list`, `get`, `create`, `update`, `delete`	`/workflow`, `/workflow/{id}`
`truto_api.workflow_runs`	`list`, `get`, `delete`	`/workflow-run`, `/workflow-run/{id}`

Datastores

Accessible via truto_api.datastores. API reference.

Method	Endpoint
`list`, `get`, `create`, `update`, `delete`	`/datastore`, `/datastore/{id}`
`test(id, method, body=None)`	`POST /datastore/{id}/test/{method}`

Static gates

Accessible via truto_api.static_gates. API reference.

Method	Endpoint
`list`, `get`, `create`, `update`, `delete`	`/static-gate`, `/static-gate/{id}`

Observability destinations

Accessible via truto_api.observability_destinations. API reference.

await truto_api.observability_destinations.test("observability-destination-id")

Method	Endpoint
`list`, `get`, `create`, `update`, `delete`	`/observability-destination`, `/observability-destination/{id}`
`test(id)`	`POST /observability-destination/{id}/test`

Documentation

Accessible via truto_api.documentations. API reference.

Method	Endpoint
`list`, `get`, `create`, `update`, `delete`	`/documentation`, `/documentation/{id}`

Batch requests

Accessible via truto_api.batch_requests. API reference.

Method	Endpoint
`create(data)`	`POST /batch-request`

Analytics

Accessible via truto_api.analytics. API reference.

Query parameters are keyword arguments. For names that aren't valid Python identifiers, such as from, pass a dict as the first argument.

dashboard = await truto_api.analytics.dashboard(range="7d")
series = await truto_api.analytics.timeseries(
    {"from": "2026-10-01", "to": "2026-10-08"},
    environment_id="environment-id",
    metric="api_calls",
)
by_integration = await truto_api.analytics.breakdown(
    environment_id="environment-id", metric="api_calls", dimension="integration"
)

Method	Endpoint
`status(...)`	`GET /analytics/status`
`metrics(...)`	`GET /analytics/metrics`
`dashboard(...)`	`GET /analytics/dashboard`
`timeseries(...)`	`GET /analytics/timeseries`
`breakdown(...)`	`GET /analytics/breakdown`

Environments, API tokens, categories, and logs

Attribute	Methods	Endpoint	Reference
`truto_api.environments`	`list`, `get`	`/environment`	Environments
`truto_api.api_tokens`	`list`, `get`	`/api-token`	API tokens
`truto_api.categories`	`list`	`/category`	Categories
`truto_api.logs`	`list`	`/log`	Logs

truto_api.ping() calls GET /ping to check that your token works.

Other resources

These resources are in the SDK but not in the public API reference.

Attribute	Methods	Endpoint
`truto_api.users`	`list`, `get`	`/user`, `/user/{id}`
`truto_api.teams`	`list`, `get`	`/team`, `/team/{id}`
`truto_api.daemons`	`list`, `get`, `create`, `update`, `delete`	`/daemon`, `/daemon/{id}`
`truto_api.daemon_jobs`	`list`, `get`, `create`, `update`, `delete`	`/daemon-job`, `/daemon-job/{id}`
`truto_api.daemon_job_runs`	`list`, `get`, `create`, `update`, `delete`	`/daemon-job-run`, `/daemon-job-run/{id}`
`truto_api.daemon_job_cron_triggers`	`list`, `get`, `create`, `update`, `delete`, `schedule`	`/daemon-job-cron-trigger`, `/daemon-job-cron-trigger/{id}`
`truto_api.sandbox_integrated_accounts`	`list`, `get`, `create`	`/sandbox-integrated-account`, `/sandbox-integrated-account/{id}`
`truto_api.integrated_account_webhooks`	`process(integrated_account_id, payload)`	`POST /integrated-account-webhook/{id}`
`truto_api.environment_integration_webhooks`	`process(environment_integration_id, payload)`	`POST /environment-integration-webhook/{id}`

Requirements