SIM Swap Detection

Detect recent SIM swap events on Safaricom subscriber numbers to prevent account takeover and fraud during high-value transactions.

⚠️Paid API & Mandatory Onboarding

Unlike standard M-Pesa APIs (like STK Push or C2B), the IMSI checkATI / SIM Swap API is a paid service on the Safaricom Developer Portal (Daraja).

Requirements before integration:

You must request access and complete commercial onboarding with Safaricom to enable the API on your App credentials.
In production, calls to this endpoint consume API credits or billed usage per query according to your Safaricom tariff plan.

Environment Variables Setup

To keep credentials secure, configure your consumer key, secret, and target phone numbers in your .env file or environment settings:

.env

# Safaricom Daraja Credentials (Must have Swap/IMSI API product enabled)
MPESA_CONSUMER_KEY=your_consumer_key_here
MPESA_CONSUMER_SECRET=your_consumer_secret_here

# M-Pesa Environment: "sandbox" or "production"
MPESA_ENVIRONMENT=sandbox

# Optional: Default target subscriber for testing
TEST_CUSTOMER_NUMBER=0722000000

Parameters Definition

Parameter	Type	Description
customerNumberrequired str	String	The subscriber phone number (e.g. '0722000000', '254722000000', or '+254722000000'). Automatically normalized by SwapRequest.
http_clientrequired HttpClient / AsyncHttpClient	Object	An instance of mpesakit's HTTP client configured for sandbox or production.
token_managerrequired TokenManager / AsyncTokenManager	Object	Instance of TokenManager managing OAuth access tokens.

Overview & Architecture

The Swap and AsyncSwap classes interface with Safaricom's /imsi/v2/checkATI endpoint to check if a subscriber's SIM card has been swapped recently.

Automatic MSISDN Normalization: Normalizes local formats (e.g. 0722... or +254722...) into standard 254722... format via SwapRequest.
Smart Fraud Helper: Exposes response.is_recently_swapped, which automatically checks if lastSwapDate differs from the default baseline (01-01-1900 00:00).
Sync & Async Support: Ready for synchronous web frameworks (Django/Flask) or asynchronous engines (FastAPI/Tornado).

Synchronous Usage Example

Swap (Synchronous)

import os
from dotenv import load_dotenv
from mpesakit.auth import TokenManager
from mpesakit.http_client import MpesaHttpClient
from mpesakit.swap import Swap, SwapRequest

load_dotenv()

env = os.getenv("MPESA_ENVIRONMENT", "sandbox")

# 1. Initialize HTTP client and TokenManager
http_client = MpesaHttpClient(env=env)
token_mgr = TokenManager(
  consumer_key=os.getenv("MPESA_CONSUMER_KEY"),
  consumer_secret=os.getenv("MPESA_CONSUMER_SECRET"),
  http_client=http_client,
)

# 2. Instantiate Swap service
swap_service = Swap(
  http_client=http_client,
  token_manager=token_mgr,
)

# 3. Create request and perform query
request = SwapRequest(customerNumber=os.getenv("TEST_CUSTOMER_NUMBER", "0722000000"))
response = swap_service.swap_request(request)

if response.is_recently_swapped:
  print(f"FRAUD WARNING: SIM swapped on {response.lastSwapDate}")
else:
  print("SAFE: No recent SIM swap detected")

Asynchronous Usage Example

AsyncSwap (Asynchronous)

import os
import asyncio
from dotenv import load_dotenv
from mpesakit.auth import AsyncTokenManager
from mpesakit.http_client import MpesaAsyncHttpClient
from mpesakit.swap import AsyncSwap, SwapRequest

load_dotenv()

async def check_sim_swap():
  env = os.getenv("MPESA_ENVIRONMENT", "sandbox")

  http_client = MpesaAsyncHttpClient(env=env)
  token_mgr = AsyncTokenManager(
      consumer_key=os.getenv("MPESA_CONSUMER_KEY"),
      consumer_secret=os.getenv("MPESA_CONSUMER_SECRET"),
      http_client=http_client,
  )

  swap_service = AsyncSwap(
      http_client=http_client,
      token_manager=token_mgr,
  )

  request = SwapRequest(customerNumber=os.getenv("TEST_CUSTOMER_NUMBER", "0722000000"))
  response = await swap_service.swap_request(request)

  print(f"Request Ref ID: {response.requestRefID}")
  print(f"Last Swap Date: {response.lastSwapDate}")
  print(f"Is Recently Swapped: {response.is_recently_swapped}")

asyncio.run(check_sim_swap())

Response Model (SwapResponse)

Parameter	Type	Description
requestRefIDrequired str	String	Unique transaction query ID generated by Safaricom.
responseCoderequired str	String	API return code ('200' for successful check).
responseDescrequired str	String	Description from Safaricom API (e.g. 'Success').
lastSwapDaterequired str	String	Timestamp of last swap in 'DD-MM-YYYY HH:MM' format. Baseline default is '01-01-1900 00:00'.
is_recently_swapped bool (property)	Boolean	Evaluates to True if lastSwapDate is strictly greater than '01-01-1900 00:00'.

💡Endpoints Used

Sandbox: https://sandbox.safaricom.co.ke/imsi/v2/checkATI
Production: https://api.safaricom.co.ke/imsi/v2/checkATI

Auth & Token Management - Managing credentials and OAuth tokens
Getting Credentials - How to set up Daraja portal apps
Production Checklist - Going live on Safaricom M-Pesa APIs