tukupay

Official Python library for the TukuPay API: register users, give them wallets, collect and send money over M-Pesa, and send SMS, from your own server.

The library handles authentication, errors and safe retries for you, and returns TukuPay's responses unchanged, as plain dicts.

Beta. The interface may still change before 1.0. Pin the version you install.

Contents

• Install
• Before you start
• How it works
• Quick start
• Guide
  1. Register a user
  2. Sign in and keep the session
  3. Give the user a wallet
  4. Set the payment PIN
  5. Collect money by M-Pesa
  6. Send money
  7. Find out what happened
  8. See everything your app has done
  9. Send SMS
• Errors
• Retries, timeouts and rate limits
• Configuration
• Reference
• Troubleshooting

Install

pip install --pre tukupay

Requires Python 3.9 or later. Its only dependency is httpx.

--pre is needed while the library is in beta; without it pip skips pre-release versions.

Before you start

TukuPay runs two separate environments. They share nothing: each has its own apps, users, wallets and API keys.

To use an environment you need, on that environment:

1. A TukuPay app. Your integration is an app. Users you register belong to it, and payments made through it are recorded against it.
2. An API key for that app. Keys are issued per app from your TukuPay account; if you do not have one yet, ask your TukuPay contact. The full key is shown once, so store it in an environment variable or secret manager.

A key only exists on the environment that issued it. The library reads the prefix and talks to the matching server, so going live is a matter of creating a sk_live_ key on production and changing one environment variable:

TUKUPAY_API_KEY=sk_test_...   # testing and development
TUKUPAY_API_KEY=sk_live_...   # production

Users, wallets and ids do not carry over: a user registered on the test environment does not exist on production.

The test environment is not a sandbox. A payout made with a test key moves real money. When testing, only pay phone numbers you control, and use small amounts.

How it works

Every request carries up to two credentials. Knowing which is which explains the whole library.

• With the key alone you can read things about your app as a whole: its wallets, transactions and users, and send SMS. These live under tuku.app.
• To act as a user (open a wallet, pay, upload KYC) you need the key and that user's token, plus their PIN whenever money leaves a wallet. These live on the object returned by tuku.as_user(tokens).

Your users never see the API key. They register and sign in with their phone number and password as on TukuPay itself; your server passes those to TukuPay and keeps the tokens it gets back.

Never put the API key in a web page, a mobile app or a public repository. Anyone holding it can act as your app.

Quick start

Check that your key works:

import os
from tukupay import TukuPay

tuku = TukuPay(os.environ["TUKUPAY_API_KEY"])

app = tuku.app.retrieve()
print(app["name"], app["status"], app["environment"])
# My Shop  active  test

If this prints your app's name, the key is valid and you are talking to the right environment.

Guide

The steps below take one user from nothing to a completed payment. Each builds on the one before.

1. Register a user

tuku.auth.register(
    phone_number="0712345678",
    password="a-strong-password",
    accepted_terms=True,
)

• Leave app_id out. The user is placed under your app because the request carries your key.
• phone_number accepts 07XXXXXXXX, 01XXXXXXXX or 2547XXXXXXXX.
• accepted_terms must be true; the user has to have agreed to TukuPay's terms.

TukuPay texts the user a one-time code. Ask them for it and confirm their phone:

tuku.auth.verify_phone(phone_number="0712345678", otp_code="123456")

A user cannot sign in until their phone is verified. If the code did not arrive, send another with tuku.auth.send_phone_verification(phone_number).

2. Sign in and keep the session

tokens = tuku.auth.login(phone_number="0712345678", password="a-strong-password")

def save_tokens(fresh):
    # Called whenever the library renews the tokens. Save them.
    db.save_tokens(user_id, fresh["access_token"], fresh["refresh_token"])

user = tuku.as_user(tokens, on_token_refresh=save_tokens)

me = user.account.me()

login returns two tokens:

You do not manage this by hand. When a call fails because the access token has expired, the library exchanges the refresh token, repeats the call once, and gives you the new pair through the callback. Store the new pair every time: the old refresh token no longer works, so a stale one means the user has to sign in again.

Sign in once and reuse the tokens; signing a user in on every request will hit TukuPay's limit on sign-ins per hour. On a later request, rebuild the user from the tokens you stored:

saved = db.load_tokens(user_id)

user = tuku.as_user(
    access_token=saved.access_token,
    refresh_token=saved.refresh_token,
    on_token_refresh=save_tokens,
)

3. Give the user a wallet

A new user has no wallet. Opening one is a three-step identity check.

Step 1: send their details. TukuPay texts the user an OTP.

user.onboarding.initial(
    first_name="John",
    middle_name="K",
    last_name="Doe",
    country_code="254",
    document_type="1",
    document_number="12345678",
    email="john@example.com",
)

document_type is "1" for a national ID, "2" for a passport and "3" for an alien ID.

Step 2: confirm with the OTP.

user.onboarding.confirm(otp="123456")

A wrong or expired OTP raises GatewayError; the message says why.

Step 3: upload the identity documents.

with open("photo.jpg", "rb") as photo, open("id-front.jpg", "rb") as front, open("id-back.jpg", "rb") as back:
    user.onboarding.kyc(files={
        "passport_size_photo": ("photo.jpg", photo, "image/jpeg"),
        "document_image_front": ("id-front.jpg", front, "image/jpeg"),
        "document_image_back": ("id-back.jpg", back, "image/jpeg"),
    })

Check progress at any time:

status = user.onboarding.status()["status"]

Once the wallet exists you can read it:

wallets = user.wallets.list()
wallet = wallets[0]
print(wallet["id"], wallet["alias"], wallet["balance"])

4. Set the payment PIN

Every payment that takes money out of a wallet needs the user's PIN (above KES 50,000, a one-time code sent to their phone instead; see step 6). Set it once:

user.account.setup_pin(pin="1234", confirm_pin="1234")

The library passes PINs straight to TukuPay and never stores or logs them. Do not store them either; ask the user each time.

5. Collect money by M-Pesa

This sends an M-Pesa prompt to a phone and puts the money in the user's wallet.

payment = user.payments.direct({
    "transaction_data": {
        "transaction_type": "WALLET_TOPUP",
        "amount": 150,
        "destination_wallet_id": wallet["id"],
    },
    "payment_source": {
        "payment_method": "mpesa",
        "phone_number": "0712345678",      # the phone that receives the M-Pesa prompt
    },
    "description": "Order 1042",
})

print(payment["transaction_id"], payment["status"])
instructions = (payment.get("type_result") or {}).get("customer_message")   # show this to the payer

Two things to do with the response:

• Show the payer type_result.customer_message word for word, if present. Some payers do not receive the M-Pesa prompt, and this text gives them the PayBill steps to pay manually.
• Do not treat this as paid. A successful response means the request was accepted. The payer still has to enter their M-Pesa PIN. Go to step 7 for the outcome.

No TukuPay PIN is needed here, because money is coming in.

6. Send money

This pays from the user's wallet to an M-Pesa number. Money is leaving a wallet, so the PIN is required.

payout = user.payments.direct({
    "transaction_data": {
        "transaction_type": "USER_TRANSFER",
        "amount": 500,
        "transfer_method": "mpesa",
        "destination_phone": "0798765432",
    },
    "payment_source": {
        "payment_method": "tukupay_wallet",
        "source_wallet_id": wallet["id"],
    },
    "pin": "1234",                         # asked from the user, never stored
})

Check first, if you want to show fees or catch problems early. validate takes the same payment and reports fees, limits and whether the balance covers it, without moving money:

check = user.payments.validate(payment_body)

Above KES 50,000 a one-time code replaces the PIN. Request the code, collect it from the user, and send the payment with otp_code and the payment_hash you were given, in place of pin:

otp = user.payments.request_otp(amount=75_000, transaction_type="USER_TRANSFER", source_wallet_id=wallet["id"])

user.payments.direct({**payment_body, "otp_code": "654321", "payment_hash": otp["payment_hash"]})

Other payment types use the same call with a different transaction_data:

user.payments.transaction_types() lists every type available to you with a short description. For a type or field not covered here, ask your TukuPay contact. To pay many M-Pesa numbers from one wallet in one call, use user.payments.bulk_transfer().

Paying twice by accident

Networks fail at the worst moment: your request may reach TukuPay while the reply never reaches you. To make a repeat safe, every payment carries an idempotency key. TukuPay remembers it, and a second request with the same key returns the first result without paying again.

• The library generates one per call and reuses it for its own automatic retries. You get this for free.
• If your code may retry (a queue job that can run twice, a user double-clicking), pass your own key built from something stable such as your order id:

user.payments.direct(payment_body, idempotency_key=f"order-{order.id}-attempt-1")

• A deliberate new attempt after a failure is a new payment and needs a new key. Reusing the old key returns the old failure.

7. Find out what happened

A payment starts as pending and ends as completed or failed. wait reads it until it settles:

from tukupay import WaitTimeoutError, failure_of

try:
    transaction = user.transactions.wait(
        payment["transaction_id"],
        timeout=120,     # give up after 2 minutes
        interval=3,      # check every 3 seconds
    )
except WaitTimeoutError:
    ...  # Still pending. Keep the id and check again later.
else:
    if transaction["status"] == "completed":
        mark_order_paid(order.id)
    else:
        failure = failure_of(transaction) or {}
        print(failure.get("category"), failure.get("hint"))
        # cancelled_by_payer  The payer cancelled the STK prompt on their handset.

wait returns for both outcomes, so always check status. If time runs out it raises WaitTimeoutError; the payment may still complete later, so keep the transaction id and check again rather than assuming failure.

wait blocks while it polls. In a web handler it is usually better to return to the user straight away and check the outcome from a background task or when they reload the page, using transactions.get(id).

When a payment fails, failure_of(transaction) gives the reason in a form you can act on:

8. See everything your app has done

These calls need only the API key and cover every user of your app. They suit back-office screens, reconciliation and reports.

from datetime import date

wallets = tuku.app.wallets.list(limit=50)
wallet = tuku.app.wallets.get(28)

page = tuku.app.transactions.list(status="completed", start_date=date(2026, 10, 1), limit=100)
print(page["total"], page["has_more"])

transaction = tuku.app.transactions.get(1042)
users = tuku.app.users.list(limit=20)["users"]

Lists return at most 100 rows per call. To walk a whole list without handling pages yourself:

for transaction in tuku.app.transactions.iterate(status="completed"):
    print(transaction["id"], transaction["amount"])

Transaction filters: transaction_type, status (pending, completed, failed), wallet_id, start_date, end_date, limit, offset.

Your key only ever sees your own app. Asking for another app's wallet or transaction gives NotFoundError, the same as one that does not exist.

9. Send SMS

Messages go out on the app owner's TukuPay account and use the owner's SMS credits.

queued = tuku.app.sms.send(
    phone_numbers=["0712345678", "0798765432"],
    message="Your order has shipped.",
)

outbox = tuku.app.sms.outbox(limit=20)
one = tuku.app.sms.status(outbox[0]["id"])
print(one["phone_number"], one["status"])

• send queues the message. The message_id it returns identifies the queued batch, not the individual texts.
• To see what was delivered, read the outbox. Each row is one recipient, with a status of success, failed or pending.
• Credits are taken when the message is actually sent, not when it is queued.
• A custom sender_id must be one approved on the owner's account.

Errors

Every failure raises a TukuPayError or one of its subclasses. Each carries:

• str(error): a readable message.
• error.status: the HTTP status, or None if no response arrived.
• error.errors: exactly what the API sent in errors. For a validation failure this is a list with one entry per bad field.
• error.body: the whole response body.

from tukupay import RateLimitError, TukuPayError, ValidationError

try:
    user.payments.direct(payment_body)
except ValidationError as error:
    show_to_user(str(error))               # "amount: Input should be greater than 0"
except RateLimitError as error:
    retry_in(error.retry_after)
except TukuPayError as error:
    log(error.status, error.errors)

Retries, timeouts and rate limits

Retries. A failed call is retried up to twice, waiting a little longer each time.

Timeouts. Each request is given 30 seconds by default.

Rate limits. Limits are per key, per minute. Every response carries X-RateLimit-Limit, X-RateLimit-Remaining and X-RateLimit-Reset. If you are regularly limited, spread the work out or ask TukuPay to raise the limit on your key.

Configuration

tuku = TukuPay(
    os.environ["TUKUPAY_API_KEY"],
    timeout=30.0,
    max_retries=2,
    debug=True,
)

The client holds a connection pool. Create it once and reuse it; call tuku.close() on shutdown, or use it as a context manager (with TukuPay(...) as tuku:).

Reference

Every call returns the API's JSON as a dict or list. The client is synchronous and safe to share between threads; an async client is planned.

tuku.app: the API key alone

tuku.auth: the API key alone

As a user: key and token

Obtained from tuku.as_user(tokens).

Helpers

Troubleshooting

Invalid or revoked API key. One of:

• The key is mistyped or was revoked.
• It was rotated more than 24 hours ago, so the old key has expired.
• It belongs to the other environment. A sk_test_ key is unknown to production and a sk_live_ key is unknown to the test environment. Remove any base URL override so the library picks the server from the key.

App not approved or not found. Your app has not been approved yet or has been suspended. Contact TukuPay.

Please verify your phone number before logging in. Finish step 1: confirm the OTP.

Too many login attempts. TukuPay allows a limited number of sign-ins per user per hour (about 10). Sign a user in once, store the tokens and reuse them; do not sign in on every request.

The user is asked to sign in again unexpectedly. A refresh token was used twice. Make sure the token callback saves the new pair, and that two servers are not refreshing the same user from the same old token.

Incorrect PIN or PIN is locked. The user entered the wrong PIN. Ask them again. After several wrong attempts the PIN locks for a while, so never retry a payment automatically after this error.

A payment stays pending. The payer has not responded to the M-Pesa prompt, or the gateway has not reported back yet. Keep checking the same transaction; do not start a second payment for the same order.

I retried and got the same failure. You reused the idempotency key. A new attempt needs a new key.

Versioning

This library follows semantic versioning. While it is 0.x, a minor release may change the interface; see CHANGELOG.md. It ignores response fields it does not know, so additions to the API do not break it.

Support

Report problems at https://github.com/tukupay/tukupay-sdk/issues. Never include an API key, token or PIN in an issue.

License

MIT