Python SDK
Add partner reporting and operational workflows to a Python service.
Install
Download thesauros-1.1.0-py3-none-any.whl from SDK downloads:
python -m pip install ./thesauros-1.1.0-py3-none-any.whl
This review release has not been published to PyPI. Python 3.9+ is required. The synchronous client uses the standard library and has no runtime dependencies. In an async service, use your worker/thread execution path for blocking SDK calls.
Connect to the Partner API
import os
from thesauros import PartnerClient
client = PartnerClient(
api_key=os.environ['THESAUROS_API_KEY'],
base_url=os.environ['THESAUROS_API_BASE'],
)
summary = client.partner.summary()
print(summary['partner']['name'], summary['as_of'])
history = client.rates.history('USDC')
print(history['scope'], history['observations'])
Use the onboarding URL, including /api/v1, and a partner-bound key. An explicit URL is required. Partner asset identifiers are USDC and USDT0; the sandbox uses USDC and USDT.
Attribute a customer
Partner write bodies are dictionaries. Query options are keyword arguments.
user = client.users.create({
'external_id': 'your-partner:customer-1042',
'wallets': ['0x1111111111111111111111111111111111111111'],
})
positions = client.partner.user_positions(user['id'])
activity = client.users.ledger(user['id'], limit=50)
Replace the example wallet with the customer's address. Namespace the globally unique external ID and retain the returned user ID. Creation is not an upsert. These calls create attribution and read records; they do not move customer assets.
Partner resources
| Resource | Methods |
|---|---|
partner |
summary, users, deposits, withdrawals, tvl, earnings, points, revenue, user_positions |
rates |
history |
vaults |
list, history |
users |
create, ledger |
analytics |
signals, regime, uplift, decisions, advisor |
reconciliation |
balances, ledger, snapshots, report |
webhooks |
create, list, event_types, deliveries, update, delete, test |
usage, status |
get |
keys |
create, list, revoke |
partners |
create, list, retrieve, update |
campaigns |
create, list, update |
partner.yield_history is retained for compatibility; prefer rates.history. Administrative methods require their own scopes. Python query names that collide with keywords use a trailing underscore, such as from_ on snapshots. See the method reference.
Prototype with the sandbox
from thesauros import SandboxClient
sandbox = SandboxClient(
api_key=os.environ['THESAUROS_SANDBOX_KEY'],
base_url=os.environ['THESAUROS_SANDBOX_BASE'],
)
position = sandbox.positions.create(
wallet='0x1111111111111111111111111111111111111111',
asset='USDC', amount=1000,
)
sandbox.positions.withdraw(position['id'], all=True)
Sandbox writes retain their original keyword-argument API. Positions are simulated. The original Thesauros class remains available as the sandbox client. Access sandbox rates through yield_ or its rates alias.
Responses and errors
Methods return data. client.last_meta holds envelope metadata; client.last_response holds the latest HTTP status, request ID and rate-limit values. Use one client per concurrent workflow when associating metadata with a specific call.
from thesauros import ApiError, NetworkError, RateLimitError
try:
rows = client.reconciliation.ledger(limit=50, offset=0)
print(len(rows), client.last_meta)
except RateLimitError as error:
print(error.retry_after, error.request_id)
except ApiError as error:
print(error.status, error.code, error.request_id)
except NetworkError as error:
print(str(error))
Pagination is explicit. The client does not load additional pages automatically.
Transport configuration
timeout defaults to 30 seconds for blocking socket operations, including body reads. max_retries defaults to 3 additional attempts. Only GET requests retry 429 and 5xx with backoff and server retry hints. Writes and network failures are not replayed automatically. Reconcile an uncertain write before sending it again.
Redirects are rejected. Malformed successful responses raise ThesaurosError. The base URL must be absolute HTTP(S) without embedded credentials, query or fragment. Socket timeout is not a total wall-clock deadline for a response that continues delivering data; retry waits are additional time.
Verify webhook deliveries
from thesauros import verify_signature
def valid_delivery(secret, headers, raw_body):
return verify_signature(
secret, headers.get('Webhook-Signature'), raw_body,
tolerance_seconds=300,
)
Pass the unchanged body bytes. The helper verifies HMAC-SHA256 and optionally timestamp age. Persist and deduplicate verified event IDs before applying effects; the helper does not store replay state. webhooks.test sends a real HTTP request to the registered receiver.
Build from source
In sdk/python:
python -m pip wheel --no-deps --wheel-dir dist .
PYTHONPATH=. python -m unittest discover -s tests -v
The backend's SDK integration suite also checks all 43 Partner methods through this client against a local database. Partner TypedDict models are generated from the verified contract and available in thesauros.partner_types. The package includes py.typed. License: MIT.