---

The official Python library for the Tungo Platform API (https://api.tungo.kz): the AI accountant, 1C integration, bank statements, e-invoices, tax forms and automations of Tungo AI.

It is generated from the API's OpenAPI contract, so its methods and models always match the contract it was built from.

Requirements

Python 3.10 or newer. The library depends on httpx and pydantic 2 (plus anyio and typing-extensions, which those two already bring in). It ships type information (py.typed) and is checked with mypy --strict and pyright in strict mode.

Installation

pip install tungoai

The import package has the same name: import tungoai.

Quick start

from tungoai import Tungo

with Tungo() as client:  # reads TUNGO_API_KEY from the environment
    companies = client.companies.list()
    for company in companies.data:
        print(company.id, company.name)

    turn = client.ask(companies.data[0].id, "How much VAT is due this quarter?")
    print(turn.answer)

Authentication and configuration

Create one client and share it: it is immutable and safe to use from several threads or tasks. Every option can be passed explicitly; three of them fall back to environment variables.

Option	Environment variable	Default
API key	TUNGO_API_KEY	none
Base URL	TUNGO_BASE_URL	https://api.tungo.kz
Webhook secret	TUNGO_WEBHOOK_SECRET	none
Timeout		60 s (600 s for non-streaming chat turns)
Maximum retries		2
Default headers, HTTP client		none, built-in

The API key is sent only to operations that require it; public endpoints and incoming hooks never receive it, and it never appears in error messages or debug output. Calling an authenticated operation without a key fails locally, before any request is made.

Every method also accepts per-request options: timeout, maximum retries, extra headers, an idempotency key and cancellation.

import os

import httpx

from tungoai import Tungo

http_client = httpx.Client(limits=httpx.Limits(max_connections=10))  # proxies, mTLS, custom transports
client = Tungo(
    api_key=os.environ["TUNGO_API_KEY"],  # default: the TUNGO_API_KEY environment variable
    base_url=os.environ.get("TUNGO_BASE_URL"),  # default: TUNGO_BASE_URL, then https://api.tungo.kz
    webhook_secret=os.environ.get("TUNGO_WEBHOOK_SECRET"),  # default: TUNGO_WEBHOOK_SECRET
    timeout=30.0,  # seconds for a whole ordinary request; default 60
    long_running_timeout=900.0,  # seconds for a non-streaming chat turn; default 600
    max_retries=3,  # default 2
    default_headers={"X-App-Name": "billing"},
    http_client=http_client,
)
company_id = client.companies.list().data[0].id

# Options of one request.
company = client.companies.get(
    company_id,
    request_options={"timeout": 5.0, "max_retries": 0, "headers": {"X-Trace-Id": "trace-1"}},
)

# A copy of the client with other settings; it shares the connection pool.
patient = client.with_options(timeout=120.0)
print(patient.companies.get(company_id).name)

# Status, headers, request id and rate limit of a response.
response = client.companies.with_raw_response.get(company_id)
print(response.status_code, response.request_id, response.rate_limit.remaining)
print(response.parse().name)

client.close()
http_client.close()  # the SDK never closes a client it did not create

Tungo is the synchronous client and AsyncTungo the asynchronous one. Both accept the same options (http_client is an httpx.Client or an httpx.AsyncClient), are immutable after construction and release their connection pool with close() / aclose() or a with block. In asynchronous code a call is cancelled the usual way, by cancelling its task; the timeout stays in force.

timeout covers ordinary calls and long_running_timeout the non-streaming chat turns (chat.create, chat.answer, ask): setting timeout=30 does not cut a chat turn to 30 s. A timeout in request_options wins over both. A malformed API key (a line break, a space, a control or non-ASCII character inside it) is refused when the client is created, with a ConfigError that does not show the key.

Errors

Every failure is a subtype of the SDK's base error. An HTTP error response becomes an API error chosen by the error.type field of the response body, falling back to the HTTP status when the body is not the API's error envelope:

error.type	Status
invalid_request_error	400
authentication_error	401
payment_required_error	402
permission_error	403
not_found_error	404
conflict_error	409
gone_error	410
limit_error	429
api_error	500
unavailable_error	503

An API error carries the status, type, code, message, request id, field errors, the parsed retry-after delay, the response headers and the raw body. Problems that happen without a response have their own types: connection error, timeout, decode error (a successful response whose body is not valid), and configuration or validation error (raised before any request).

import tungoai
from tungoai import Tungo

client = Tungo()
try:
    client.companies.get("co_missing")
except tungoai.NotFoundError as error:
    print(error.status, error.type, error.code, error.message, error.request_id)
except tungoai.InvalidRequestError as error:
    for item in error.fields:
        print(item.path, item.code, item.message)
except tungoai.LimitError as error:
    print("retry after", error.retry_after, "s; the limit resets at", error.extra.get("resets_at"))
except tungoai.APIError as error:  # any other error response
    print(error.status, error.headers.get("x-request-id"), error.body)
except tungoai.APITimeoutError:
    print("the request timed out")
except tungoai.APIConnectionError as error:
    print("network problem:", error.__cause__)
except tungoai.DecodeError as error:
    print("unexpected response body:", error.body)
except tungoai.ConfigError as error:  # raised before any request, e.g. no API key
    print(error)
finally:
    client.close()

Every error derives from tungoai.TungoError. APITimeoutError and WaitTimeoutError also derive from the built-in TimeoutError; RequestValidationError and MoneyError from ValueError. The original exception, when there is one, is available as __cause__.

An error that a chat turn reports inside a successful response (an error event of a stream, or errors of ask) is raised by the helpers that collect a turn as ChatStreamError. It is not an APIError and has no HTTP status; its type is the same error.type (limit_error, api_error, ...), next to code, message, request_id, extra, and result or turn with what was received. The TypeScript SDK throws the APIError subclass of that error.type (for example LimitError) with an undefined status for the same situation.

Retries and timeouts

A request is retried up to 2 times (3 attempts) on connection errors, timeouts and the statuses 408, 429, 500, 502, 503 and 504, but only when repeating it is safe: GET, HEAD, PUT and DELETE requests, and POST or PATCH requests that carry an Idempotency-Key. For the operations whose contract declares that header the SDK generates a random key and reuses it for every attempt; you can pass your own.

The delay between attempts is min(0.5 s × 2^attempt, 8 s) with random jitter. A positive retry-after-ms or Retry-After response header overrides it; if the server asks for more than 60 s the SDK does not wait and raises the error with the delay filled in.

The timeout covers the whole request including reading the body. Streaming calls have no total timeout, only a connect timeout and an idle timeout of 120 s. Cancellation supplied by the caller is combined with the timeout.

from tungoai import Tungo

client = Tungo(max_retries=5, timeout=20.0)  # for every request of this client
company_id = client.companies.list().data[0].id
proposal_id = client.proposals.list(company_id).data[0].id

# For one request. approve declares Idempotency-Key, so the SDK would generate a key anyway;
# pass your own to make a repeated call from your side safe too.
approved = client.proposals.approve(
    company_id,
    proposal_id,
    note="Checked against the contract",
    request_options={"idempotency_key": f"approve-{proposal_id}", "max_retries": 1, "timeout": 10.0},
)
print(approved.to_dict())

# A POST that does not declare the header is never retried unless you give it a key.
client.proposals.reject(
    company_id, proposal_id, reason="Duplicate", request_options={"idempotency_key": f"reject-{proposal_id}"}
)

client.close()

A request is retried only when repeating it is safe: GET, HEAD, PUT and DELETE, or any request that carries an Idempotency-Key (generated for the operations that declare it, passed as idempotency_key, or put into the headers by you). A Retry-After of zero, or a date that has already passed, does not trigger an immediate retry: the computed backoff applies.

The timeout is one deadline for the whole request. The asynchronous client enforces it exactly. The synchronous client gives each phase (connecting, sending, waiting for the headers, reading the body) only the time that is left and checks the deadline after every chunk of the body; a server that goes silent in the middle of a body is cut off when the time that was left at the start of the body runs out.

Pagination

List operations use cursor pagination ({ data, next_cursor }). The plain method returns one typed page that knows whether there is a next page and can fetch it. The auto-paging variant is a lazy iterator: it requests the next page only when needed, carries your filters over, and stops fetching as soon as you stop iterating.

from tungoai import Tungo

client = Tungo()
company_id = client.companies.list().data[0].id

# One page: the items, the cursor and the whole typed response.
page = client.proposals.list(company_id, status="awaiting_approval", limit=50)
print(len(page.data), page.next_cursor, page.body.total)
if page.has_next_page():
    page = page.get_next_page()  # same filters, next cursor

# Every item: iterating a page requests the following pages only when they are needed.
for proposal in client.proposals.list(company_id, status="awaiting_approval", limit=50):
    print(proposal.id, proposal.title, proposal.total.formatted)
    if proposal.blocker_count > 0:
        break  # no further page is requested

client.close()

A cursor that the server has already returned in the same walk (the previous one or any earlier one) raises PaginationError instead of looping forever.

Streaming

Streaming operations use server-sent events and deliver typed events as they arrive. Leaving the loop early or closing the stream closes the connection. A stream request is retried only before its first byte; an error response is raised as the same typed API error as for ordinary calls. A chat stream that ends before its final event is reported as an error, never as a finished answer.

from tungoai import ChatStreamError, DecodeError, Tungo
from tungoai.types import ChatEventDone, ChatEventText, ChatEventTool

client = Tungo()
company_id = client.companies.list().data[0].id

# Events as they arrive. Leaving the "with" block closes the connection.
with client.chat.create_stream(company_id, message="How much VAT is due this quarter?") as stream:
    print("dialog", stream.dialog_id, "turn", stream.turn_no)
    for event in stream:
        if isinstance(event, ChatEventText):
            print(event.delta, end="", flush=True)
        elif isinstance(event, ChatEventTool):
            print(f"\n[{event.name}: {event.status}]")
        elif isinstance(event, ChatEventDone):
            print("\nstop reason:", event.stop_reason)

# Or let the SDK collect the turn. An "error" event of the turn is raised as ChatStreamError,
# and a stream that was cut before its "done" event as DecodeError: never a half answer.
try:
    result = client.chat.create_stream(company_id, message="And for the year?").get_final_result()
    print(result.text, result.stop_reason, result.cost_tiyn)
except ChatStreamError as error:
    print(error.type, error.code, error.message, error.request_id)
except DecodeError as error:
    print("the stream was cut:", error.result.text if error.result else "")

client.close()

Every streaming operation of the API is available: chat.create_stream, chat.answer_stream, chat.list_events, feed.list_events and batches.list_events. Events are typed models; an event type this version does not know arrives as UnknownVariant, and heart-beats are skipped. stream.iter_sse() gives the raw server-sent events. request_options={"stream_idle_timeout": 300} changes the idle timeout of one call.

Leaving a for loop over a stream early (break, return, an exception) closes the connection: at once on CPython with the synchronous client, at the next step of the event loop with the asynchronous one. A with block, or close() / aclose(), makes the moment explicit on every interpreter. If you pull events with next(stream) instead of a loop, close the stream yourself.

get_final_result() never returns a half answer: a stream that ends without its done event raises DecodeError (error.result holds what arrived). A successful answer to a streaming call that is not text/event-stream raises DecodeError too, instead of looking like an empty stream. Plain iteration simply ends when the stream ends.

Webhooks

Outgoing webhooks are signed according to the Standard Webhooks specification (Svix). The verify helper takes the raw request body and the headers (webhook-* or svix-*), checks the HMAC-SHA256 signature in constant time and the timestamp within 300 s of the system clock, and returns the typed event. Any problem is reported as a webhook verification error. If no secret is passed, the client's webhook secret is used.

The send-hook helper posts an event to an incoming hook (/hooks/{token}) and, when given a secret, signs the exact bytes it sends in the X-Signature: sha256=<hex> header. The API key is never sent to hooks.

from collections.abc import Mapping

import tungoai
from tungoai import Tungo, WebhookVerificationError
from tungoai.types import WebhookEventProposalWritten

client = Tungo()  # the webhook secret comes from TUNGO_WEBHOOK_SECRET, or pass webhook_secret=...


def handle_webhook(raw_body: bytes, headers: Mapping[str, str]) -> int:
    """Call it from your HTTP handler with the raw request body and the request headers.

    Returns the HTTP status to answer with.
    """
    try:
        event = client.verify_webhook(raw_body, headers)
    except WebhookVerificationError as error:
        print("rejected:", error.reason)
        return 400
    if isinstance(event, WebhookEventProposalWritten):
        print("written to 1C:", event.data.proposal_id)
    return 204


# Send an event to an incoming hook. With a secret the exact bytes sent are signed (X-Signature).
accepted = client.send_hook(
    "th_example",
    {"id": "order_1532", "amount_tiyn": 4_500_000},
    secret="hksec_example",
    event_id="order_1532",
)
print(accepted.to_dict())

# Check a hook signature yourself (for example in a test double of the hook endpoint).
body = b'{"id":"order_1532"}'
assert tungoai.verify_hook_signature(body, tungoai.hook_signature_header(body, "hksec_example"), "hksec_example")

tungoai.verify_webhook(body, headers, secret) is the same check without a client. The reason of a failure is in WebhookVerificationError.reason. client.automations.post_hook and client.send_hook are the same call.

Money

The API expresses amounts in tiyn (1/100 of a tenge) as 64-bit integers. The money helper converts decimal strings and exact decimal values to tiyn without floating-point error, rounding half away from zero ("1.005" → 101), and formats tiyn for display (4500000 → 45 000,00 ₸). Invalid input raises a typed error.

from decimal import Decimal

from tungoai import MoneyError, TungoMoney, is_valid_bin_iin

print(TungoMoney.to_tiyn("45000.50"))  # 4500050
print(TungoMoney.to_tiyn(Decimal("1.005")))  # 101: half away from zero, no float error
print(TungoMoney.to_tiyn(1.005))  # 101: a float goes through its shortest decimal form
print(TungoMoney.format_tiyn(4_500_000))  # 45 000,00 ₸
print(TungoMoney.format_tiyn(-50, "USD"))  # -0,50 $

try:
    TungoMoney.to_tiyn("12,5")
except MoneyError as error:
    print(error)

print(is_valid_bin_iin("990140000017"))  # True

TungoMoney.to_tiyn accepts str, decimal.Decimal, int (whole tenge) and float. Tiyn are 64-bit integers in the API: format_tiyn raises MoneyError for a value outside that range, like every Tungo SDK.

Async usage

import asyncio

from tungoai import AsyncTungo
from tungoai.types import ChatEventText


async def main() -> None:
    async with AsyncTungo() as client:
        companies = await client.companies.list()
        company_id = companies.data[0].id

        proposals = await client.proposals.list(company_id, limit=50)
        async for proposal in proposals:  # requests the following pages as needed
            print(proposal.id, proposal.title)

        async with await client.chat.create_stream(company_id, message="What is new today?") as stream:
            async for event in stream:
                if isinstance(event, ChatEventText):
                    print(event.delta, end="")


asyncio.run(main())

Types

Responses are pydantic models from tungoai.types: read fields as attributes, model.to_dict() returns the JSON object as the API sent it. Models are forward compatible: unknown fields are kept in model.model_extra, enumerations accept values added later, and a tagged union (for example ChatEvent or WebhookEvent) yields UnknownVariant for a tag this version does not know, so narrow with isinstance. A field that is absent from the response and one that is null are both None; model.model_fields_set tells them apart.

Responses are decoded leniently, requests strictly. If a response does not match the contract (a field that must not be null is null, a field changed its type), the SDK does not fail the call: the model is built without validation, every part that does match is still a typed model, the part that does not is left as the API sent it, and to_dict() returns the original JSON. So a value can differ from its annotation after such a drift; the tungoai logger reports it at DEBUG level. The same holds for the events of a stream: one odd event does not end the turn. A body that is not JSON, or JSON of the wrong kind altogether (an array where an object is expected), still raises DecodeError.

Request parameters are keyword arguments. Leave a parameter out to not send it; pass None to send JSON null where the API allows it. Nested objects are plain dictionaries typed by tungoai.types.params.

To call an endpoint this version does not cover, build a tungoai.RequestSpec and pass it to client.request.

Helpers

client.ask sends one chat message and waits for the whole turn. client.upload_bank_statement(company_id, file_name, content) takes the bytes of the file, or a string: the base64 of the file by default, or the statement text itself with text=True (the same rule as in the TypeScript SDK). client.wait_for_run, client.run_scenario_and_wait and client.wait_for_batch poll with a pause that grows from 1 s to 5 s until the run or job is final; they raise WaitTimeoutError after timeout seconds (10 minutes by default).

Logging and secrets

The SDK logs to the tungoai logger at DEBUG level: the method, the path, the status and the request id, never header values or the API key, and the path of an incoming hook is logged as /hooks/***.

httpx, the HTTP library underneath, writes its own line for every request to the httpx logger at INFO level, with the full URL. The URL of an incoming hook contains its token, which is a secret. For the requests the SDK sends, it replaces that URL with /hooks/*** in httpx's line (a filter on the httpx logger that touches nothing else). Anything else that records URLs on its own (a custom transport, a proxy, an APM agent) is outside the SDK's reach: treat hook URLs as secrets there.

API reference

Generated from the Tungo Platform API contract: 13 resources, 82 methods, 22 webhook event types.

Resource	Description	Methods
client.workspaces	Organisations (workspaces) the API key can access	list
client.companies	Companies of an organisation and their profiles	list · get
client.platform	Platform balance, ledger, usage reports, limits, audit log and model rates	list_usage_rates · get_account · get_audit · get_ledger · list_limits · list_requests · get_usage
client.api_keys	The current API key and its usage statistics	get_current · get_current_usage · get_usage
client.webhooks	Outgoing webhooks of an organisation (Svix): endpoints, event catalogue, deliveries and signature check	list_endpoints · create_endpoint · get_endpoint · update_endpoint · delete_endpoint · list_endpoint_attempts · rotate_endpoint_secret · test_endpoint · list_event_types · list_messages · list_message_attempts · resend_message · get_usage
client.chat	Conversations with the AI accountant: questions, streaming chat (SSE), token counting and history	list · create · create_stream · count_tokens · get · update · delete · answer · answer_stream · list_events · stop
client.proposals	Documents proposed by the AI accountant: review, approval for posting to 1C, and rejection	list · approve_batch · get · approve · reject
client.feed	The company's feed of events and proposals (pages and a live SSE stream)	get · list_events
client.batches	Processing batches for bank statements, incoming e-invoices and DirectBank	list · create_bank_statement · create_esf_purchases · get · list_events · stop · get_summary · list_statements · poll_directbank
client.esf	Incoming and outgoing electronic invoices (the ESF system)	get_incoming · get_outgoing
client.forms	Tax forms (FNO 100.00 and others): building, ISNA checks and data gaps	list · build · list_data_gaps · get_isna · list_notes
client.onec	Status of the company's connection to its 1C database	get_status
client.automations	Company automations: incoming hooks, scenarios, step chains, runs and the AI autonomy level	post_hook · list_hooks · create_hook · get_hook · update_hook · delete_hook · rotate_hook · test_hook · list_runs · get_run · cancel_run · retry_run · list_scenarios · create_scenario · get_scenario · put_scenario · update_scenario · delete_scenario · run_scenario · get_state · put_state · get_autonomy · put_autonomy

Other SDKs

The same API is available in ten languages. All SDKs are generated from one contract and behave the same way.

Language	Repository	Package
TypeScript	tungo-sdk-typescript	@tungoai/sdk
Python	tungo-sdk-python	tungoai
Go	tungo-sdk-go	github.com/TungoAI/tungo-sdk-go
Rust	tungo-sdk-rust	tungo-sdk
PHP	tungo-sdk-php	tungo/sdk
Java	tungo-sdk-java	kz.tungo:tungo-sdk
Kotlin	tungo-sdk-kotlin	kz.tungo:tungo-sdk-kotlin
Swift	tungo-sdk-swift	TungoSDK
Ruby	tungo-sdk-ruby	tungo-sdk
.NET	tungo-sdk-dotnet	Tungo.Sdk

Versioning and support

This library follows Semantic Versioning. While the major version is 0, a minor release may contain breaking changes; they are always listed in the changelog. This README describes version 0.2.0.

The code is generated: please report problems and ideas in the issue tracker rather than opening pull requests against generated files. To report a vulnerability, follow the security policy.

License

MIT