Daraja Agent KitTokens
By the end of this track you can make an authenticated call and explain the async model.
0/4 finished
Step 1
The idea
Every call needs a bearer token. You get one by sending the key and secret to the OAuth endpoint, and it lasts about an hour. Good code caches it and fetches a new one on a 401.
Step 2
How it works
1. Build the Basic header
Join your consumer key and secret with a colon, then base64-encode the result. This is standard HTTP Basic authentication, and it is only used for this one call.
AUTH=$(printf "%s:%s" "$DARAJA_CONSUMER_KEY" "$DARAJA_CONSUMER_SECRET" | base64 -w0)2. Ask for a token
Send a GET to the OAuth endpoint with that header.
curl -s "https://sandbox.safaricom.co.ke/oauth/v1/generate?grant_type=client_credentials" \ -H "Authorization: Basic $AUTH"3. Read the answer
access_token is the token. expires_in is its lifetime in seconds, about an hour. Note that Daraja sends it as a string.
{ "access_token": "c9SQxWWhmdVRlyh0zh8gZDTkubVF", "expires_in": "3599"}4. Use it on every other call
Every other Daraja request sends the token as a Bearer header. The token is opaque: don't try to decode it.
Authorization: Bearer c9SQxWWhmdVRlyh0zh8gZDTkubVF5. Cache it and refresh it
Fetching a new token before every call is slow and can get you rate-limited. Keep the token in memory with the time it expires, and refresh it a minute early. If a call still returns 401 or 404.001.03 (invalid access token), clear it, fetch a new one, and retry the call once.
import base64, os, time, requestsBASE = "https://sandbox.safaricom.co.ke"_cache = {"token": None, "expires": 0.0}def get_token(): if _cache["token"] and time.time() < _cache["expires"] - 60: return _cache["token"] pair = f'{os.environ["DARAJA_CONSUMER_KEY"]}:{os.environ["DARAJA_CONSUMER_SECRET"]}' r = requests.get( f"{BASE}/oauth/v1/generate", params={"grant_type": "client_credentials"}, headers={"Authorization": "Basic " + base64.b64encode(pair.encode()).decode()}, timeout=10, ) r.raise_for_status() body = r.json() _cache["token"] = body["access_token"] _cache["expires"] = time.time() + int(body["expires_in"]) return _cache["token"]Step 3
Try it
mpesa status fetches a token for you.
With the MCP server, ask your agent to use mpesa_check_credentials
Step 4
Read the response
Success means the key and secret are right. A 400 or 401 here means wrong keys or the wrong environment.
Step 5
Common mistakes
- Treating
expires_inas a number without converting it. It arrives as a string. - Retrying a 401 in a loop. Refresh once and retry once; if it fails again, the keys or environment are wrong.
- Logging the token or the Basic header. Both let someone call Daraja as you.
Go deeper
What should your code do when a call returns 401?
Answer in your own words first.
Show the answerHide the answer
Get a new token once, and retry once.