> ## Documentation Index
> Fetch the complete documentation index at: https://docs.palpluss.com/llms.txt
> Use this file to discover all available pages before exploring further.

# Python SDK

> Use the PalPluss Python SDK with FastAPI, Django, Flask, and both sync and async patterns.

# Python SDK

```bash theme={null}
pip install palpluss
```

Requires Python 3.9+. Both synchronous and asynchronous clients are included.

***

## Client initialisation

```python theme={null}
from palpluss import PalPluss, AsyncPalPluss

# Sync client (works with Flask, Django, scripts)
client = PalPluss(
    api_key="pk_live_...",      # or reads PALPLUSS_API_KEY
    timeout=30.0,
    auto_retry_on_rate_limit=True,
    max_retries=3,
)

# Async client (works with FastAPI, async Django, etc.)
async_client = AsyncPalPluss(api_key="pk_live_...")
```

Use as a context manager to ensure the underlying HTTP session is closed cleanly:

```python theme={null}
# Sync
with PalPluss() as client:
    result = client.stk_push(amount=500, phone="254712345678")

# Async
async with AsyncPalPluss() as client:
    result = await client.stk_push(amount=500, phone="254712345678")
```

***

## Framework examples

<Tabs>
  <Tab title="FastAPI">
    Create a shared client instance at module level — do not instantiate per request:

    ```python theme={null}
    # app/palpluss.py
    from palpluss import AsyncPalPluss
    import os

    palpluss = AsyncPalPluss(api_key=os.getenv("PALPLUSS_API_KEY"))
    ```

    **Payment endpoint**

    ```python theme={null}
    # app/routers/payments.py
    from fastapi import APIRouter, HTTPException
    from pydantic import BaseModel
    from palpluss import PalPlussApiError
    from app.palpluss import palpluss

    router = APIRouter()

    class InitiatePaymentRequest(BaseModel):
        amount: float
        phone: str
        reference: str

    @router.post("/pay")
    async def initiate_payment(body: InitiatePaymentRequest):
        try:
            tx = await palpluss.stk_push(
                amount=body.amount,
                phone=body.phone,
                account_reference=body.reference,
                callback_url="https://yourapp.com/webhooks/palpluss",
                channel_id='your-payment-channel-id'
            )
            return {"transaction_id": tx["transactionId"], "status": tx["status"]}
        except PalPlussApiError as e:
            raise HTTPException(status_code=e.http_status, detail={"message": str(e), "code": e.code})
    ```

    **Webhook endpoint**

    ```python theme={null}
    # app/routers/webhooks.py
    from fastapi import APIRouter, Request, Response
    from palpluss import parse_webhook_payload
    import asyncio

    router = APIRouter()

    @router.post("/webhooks/palpluss", status_code=200)
    async def palpluss_webhook(request: Request):
        body = await request.body()

        try:
            payload = parse_webhook_payload(body.decode())
        except ValueError:
            return Response(content='{"error":"invalid payload"}', status_code=400)

        # Acknowledge before processing — return 200 immediately
        asyncio.create_task(handle_webhook(payload))
        return {"received": True}

    async def handle_webhook(payload: dict):
        event_type = payload["event_type"]
        transaction = payload["transaction"]

        if event_type == "transaction.success":
            receipt = transaction.get("mpesa_receipt")
            print(f"Payment {transaction['id']} confirmed — receipt: {receipt}")
            # fulfil order, credit account, etc.
        elif event_type == "transaction.failed":
            print(f"Payment {transaction['id']} failed: {transaction['result_desc']}")
    ```

    **App entry point**

    ```python theme={null}
    # app/main.py
    from fastapi import FastAPI
    from app.routers import payments, webhooks

    app = FastAPI()
    app.include_router(payments.router, prefix="/api")
    app.include_router(webhooks.router)
    ```
  </Tab>

  <Tab title="Django">
    Add the API key to your `.env` / settings and create a utility module:

    ```python theme={null}
    # payments/palpluss.py
    from palpluss import PalPluss
    from django.conf import settings

    _client: PalPluss | None = None

    def get_palpluss() -> PalPluss:
        global _client
        if _client is None:
            _client = PalPluss(api_key=settings.PALPLUSS_API_KEY)
        return _client
    ```

    **View**

    ```python theme={null}
    # payments/views.py
    import json
    from django.http import JsonResponse
    from django.views.decorators.csrf import csrf_exempt
    from django.views.decorators.http import require_POST
    from palpluss import PalPlussApiError, parse_webhook_payload
    from .palpluss import get_palpluss

    @require_POST
    def initiate_payment(request):
        body = json.loads(request.body)
        client = get_palpluss()

        try:
            tx = client.stk_push(
                amount=body["amount"],
                phone=body["phone"],
                account_reference=body.get("reference", ""),
                callback_url=request.build_absolute_uri("/webhooks/palpluss/"),
                channel_id='your-payment-channel-id'
            )
            return JsonResponse({"transaction_id": tx["transactionId"]})
        except PalPlussApiError as e:
            return JsonResponse({"error": str(e), "code": e.code}, status=e.http_status)


    @csrf_exempt
    @require_POST
    def palpluss_webhook(request):
        try:
            payload = parse_webhook_payload(request.body.decode())
        except ValueError:
            return JsonResponse({"error": "invalid payload"}, status=400)

        event_type = payload["event_type"]
        transaction = payload["transaction"]

        if event_type == "transaction.success":
            # fulfil order
            pass
        elif event_type == "transaction.failed":
            pass

        return JsonResponse({"received": True})
    ```

    **URLs**

    ```python theme={null}
    # payments/urls.py
    from django.urls import path
    from . import views

    urlpatterns = [
        path("api/pay/", views.initiate_payment),
        path("webhooks/palpluss/", views.palpluss_webhook),
    ]
    ```
  </Tab>

  <Tab title="Flask">
    ```python theme={null}
    # app.py
    from flask import Flask, request, jsonify
    from palpluss import PalPluss, parse_webhook_payload, PalPlussApiError
    import os
    import threading

    app = Flask(__name__)
    palpluss = PalPluss(api_key=os.getenv("PALPLUSS_API_KEY"))


    @app.post("/api/pay")
    def initiate_payment():
        body = request.get_json()

        try:
            tx = palpluss.stk_push(
                amount=body["amount"],
                phone=body["phone"],
                account_reference=body.get("reference", ""),
                callback_url=f"{os.getenv('APP_URL')}/webhooks/palpluss",
                channel_id='your-payment-channel-id'
            )
            return jsonify(transaction_id=tx["transactionId"], status=tx["status"])
        except PalPlussApiError as e:
            return jsonify(error=str(e), code=e.code), e.http_status


    @app.post("/webhooks/palpluss")
    def palpluss_webhook():
        try:
            payload = parse_webhook_payload(request.get_data(as_text=True))
        except ValueError:
            return jsonify(error="invalid payload"), 400

        # Acknowledge first — process in background thread
        threading.Thread(target=process_webhook, args=(payload,)).start()
        return jsonify(received=True), 200


    def process_webhook(payload: dict):
        if payload["event_type"] == "transaction.success":
            tx = payload["transaction"]
            print(f"Payment confirmed — receipt: {tx.get('mpesa_receipt')}")
        elif payload["event_type"] == "transaction.failed":
            tx = payload["transaction"]
            print(f"Payment failed: {tx['result_desc']}")
    ```
  </Tab>
</Tabs>

***

## STK Push

```python theme={null}
# Sync
result = client.stk_push(
    amount=500,
    phone="254712345678",
    account_reference="ORDER-001",
    transaction_desc="Payment for order #001",
    callback_url="https://yourapp.com/webhooks/palpluss",
    channel_id='your-payment-channel-id'
)
print(result["transactionId"])  # save for polling
print(result["status"])         # "PENDING"

# Async
result = await async_client.stk_push(
    amount=500,
    phone="254712345678",
    account_reference="ORDER-001",
    callback_url="https://yourapp.com/webhooks/palpluss",
    channel_id='your-payment-channel-id'
)
```

***

## B2C Payout

Always supply an `idempotency_key` so you can safely retry without double-paying:

```python theme={null}
payout = client.b2c_payout(
    amount=1000,
    phone="254712345678",
    reference="SALARY-JAN-2025-EMP001",
    description="January salary",
    idempotency_key="salary-jan-2025-emp001",
)
print(payout["transactionId"])
print(payout["status"])  # "PENDING"
```

***

## Service wallet

```python theme={null}
# Check balance
balance = client.get_service_balance()
print(balance["availableBalance"])   # e.g. 45000.0
print(balance["currency"])           # "KES"

# Top up via STK Push
topup = client.service_topup(
    amount=10000,
    phone="254712345678",
    idempotency_key="topup-jan-2025-001",
)
```

***

## Transactions

```python theme={null}
# Fetch one
tx = client.get_transaction("tx_01j8abc123")
print(tx["status"])   # "SUCCESS"

# List with pagination
cursor = None

while True:
    page = client.list_transactions(
        limit=100,
        status="SUCCESS",
        type="STK",
        cursor=cursor,
    )
    for item in page["items"]:
        print(item["transaction_id"], item["status"])

    cursor = page["next_cursor"]
    if cursor is None:
        break
```

***

## Payment wallet channels

```python theme={null}
# Create
channel = client.create_channel(
    type="PAYBILL",
    shortcode="400200",
    name="Main Paybill",
    is_default=True,
)

# Update
client.update_channel(channel["id"], name="Primary Paybill")

# Delete
client.delete_channel(channel["id"])
```

***

## Error handling

```python theme={null}
from palpluss import PalPluss, PalPlussApiError, RateLimitError

client = PalPluss(api_key="pk_live_...")

try:
    result = client.stk_push(amount=500, phone="254712345678")
except RateLimitError as e:
    # e.retry_after — seconds to wait before retrying
    print(f"Rate limited — retry after {e.retry_after}s")
except PalPlussApiError as e:
    # e.code        — machine-readable code e.g. "INVALID_PHONE"
    # e.http_status — HTTP status code       e.g. 400
    # e.request_id  — include in support requests
    print(f"[{e.code}] HTTP {e.http_status}: {e}  (request_id={e.request_id})")
```

### Common error codes

| Code                           | HTTP | Meaning                                    |
| ------------------------------ | ---- | ------------------------------------------ |
| `INVALID_PHONE`                | 400  | Phone number format not recognised         |
| `INSUFFICIENT_SERVICE_BALANCE` | 402  | Not enough service tokens to cover the fee |
| `INSUFFICIENT_FUNDS`           | 409  | B2C wallet balance below requested amount  |
| `KYC_NOT_VERIFIED`             | 403  | KYC must be approved before sending B2C    |
| `INVALID_API_KEY`              | 401  | API key missing, invalid, or revoked       |
| `RATE_LIMIT_EXCEEDED`          | 429  | 60 requests per minute limit exceeded      |

***

## Async patterns

`AsyncPalPluss` mirrors every method of `PalPluss` with the same parameters. Use it whenever you are in an `async` context:

```python theme={null}
import asyncio
from palpluss import AsyncPalPluss

async def main():
    async with AsyncPalPluss() as client:
        # Run multiple STK pushes concurrently
        results = await asyncio.gather(
            client.stk_push(amount=100, phone="254712345678"),
            client.stk_push(amount=200, phone="254787654321"),
        )
        for r in results:
            print(r["transactionId"], r["status"])

asyncio.run(main())
```