Skip to content

Python SDK

truto-python-sdk is a Python SDK to interact with the Truto API, a powerful integration platform for connecting multiple SaaS applications. The SDK mirrors the Truto REST API endpoints, which are documented in the Truto API Reference.

Requirements

Python 3.8+

Installation

pip install truto-python-sdk

Usage

import asyncio
from truto_python_sdk import TrutoApi
 
truto_api = TrutoApi(token="<your_api_token>")
 
async def main():
    # Fetch all installed integrations for the environment
    installed_integrations = truto_api.environment_integrations.list()
    async for integration in installed_integrations:
        print(integration)
 
    # Fetch integrated accounts for a tenant
    integrated_accounts = truto_api.integrated_accounts.list(tenant_id="acme-1")
    print(await integrated_accounts.to_array())
 
    # Make a request to the unified API
    unified_cursor = truto_api.unified_api.list(
        "accounting",
        "accounts",
        {
            "integrated_account_id": "766cc1ee-6637-4aa1-a73e-a0c89ccc867c",
            "created_at": "2023-05-01T00:00:00.000Z",
        },
    )
    async for account in unified_cursor:
        print(account)
 
    unified_resource = await truto_api.unified_api.get(
        "accounting",
        "accounts",
        "1",
        {"integrated_account_id": "766cc1ee-6637-4aa1-a73e-a0c89ccc867c"},
    )
    print(unified_resource)
 
asyncio.run(main())

Pagination

The SDK uses async iterators for list calls, handling pagination automatically:

cursor = truto_api.unified_api.list(
    "ticketing",
    "tickets",
    {"integrated_account_id": "c54bc595-486e-4bbb-8c17-20810fa4a86c"},
)
async for ticket in cursor:
    print(ticket)

Collect all resources with to_array():

tickets = await truto_api.unified_api.list(
    "ticketing",
    "tickets",
    {"integrated_account_id": "c54bc595-486e-4bbb-8c17-20810fa4a86c"},
).to_array()

Or fetch one page at a time with next_page():

page = await truto_api.unified_api.list(
    "ticketing",
    "tickets",
    {"integrated_account_id": "c54bc595-486e-4bbb-8c17-20810fa4a86c"},
).next_page()

Custom API

Make arbitrary HTTP requests through Truto's custom API proxy:

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

Supports get, post, put, patch, and delete methods.

MCP (Model Context Protocol)

Interact with Truto's MCP server to discover and invoke tools programmatically:

mcp_token = "your-mcp-token"
 
# Initialize an MCP session
init_response = await truto_api.mcp.initialize(
    mcp_token, "my-client", "1.0.0",
)
 
# List available tools
tools = await truto_api.mcp.list_tools(mcp_token)
 
# Call a specific tool
result = await truto_api.mcp.call_tool(mcp_token, "list_tickets", {"status": "open"})

Workflows

Create and manage automation workflows:

# List all workflows
workflows = truto_api.workflows.list()
async for workflow in workflows:
    print(workflow)
 
# Create a workflow
new_workflow = await truto_api.workflows.create({
    "trigger_name": "on_ticket_created",
    "config": {"actions": [{"type": "notify", "channel": "slack"}]},
})
 
# Update a workflow
await truto_api.workflows.update(
    "workflow-id",
    {"config": {"actions": [{"type": "notify", "channel": "email"}]}},
)

Workflow Runs

Track and manage individual executions of workflows:

runs = truto_api.workflow_runs.list(workflow_id="workflow-id")
async for run in runs:
    print(run.get("status"), run.get("result"))

Alarms

Create alarms that trigger on schedules, durations, or specific dates:

# Cron-based alarm
cron_alarm = await truto_api.alarms.create({
    "alarm_type": "cron",
    "cron_expression": "0 9 * * *",
    "entity_id": "some-entity-id",
})
 
# Duration-based alarm
duration_alarm = await truto_api.alarms.create({
    "alarm_type": "duration",
    "duration": 3600,
    "entity_id": "some-entity-id",
})

Sandbox Integrated Accounts

Create sandboxed copies of integrated accounts for testing:

sandbox = await truto_api.sandbox_integrated_accounts.create({
    "id": "766cc1ee-6637-4aa1-a73e-a0c89ccc867c",
})

Webhooks

Process incoming webhooks for integrated accounts and environment integrations:

# Integrated account webhook
await truto_api.integrated_account_webhooks.process(
    "766cc1ee-6637-4aa1-a73e-a0c89ccc867c",
    {"event": "ticket.created", "data": {"id": "123", "title": "New ticket"}},
)
 
# Environment integration webhook
await truto_api.environment_integration_webhooks.process(
    "env-integration-id",
    {"event": "sync.completed", "data": {"records": 150}},
)