How to Isolate Per-User OAuth Tokens for a Multi-User AI Agent With Python and Ollama

If your AI agent only ever acts for one person, it is easy to skip a hard question: whose access does this tool call actually use? The moment a second person starts using the same agent, that question stops being optional. A chat bot that reads a team channel and files issues on your behalf has to know, for every single call, exactly whose Slack workspace and whose GitHub repository it is allowed to touch right now. Get that wrong and the agent can quietly leak one person’s private messages to another, or file an issue under the wrong name.

In this tutorial you will build a small, real multi-provider OAuth 2.0 system from scratch in Python: two mock services standing in for a team chat app and an issue tracker, an encrypted per-user token store, an OAuth consent flow, and a local AI agent that reads messages, decides which ones describe real work, and files issues using the correct person’s own credentials. You will watch the broken, shared-token version leak data live, fix it, then prove the fix holds by running the agent as two different people and confirming neither one can touch the other’s tokens or data. Every command in this guide was run against real local servers and a real local language model while writing this post, and the output shown is copied from that session, not invented.

What You Will Build and Why It Matters

OAuth 2.0 is the standard way one application gets permission to act on a user’s behalf inside another application, without ever seeing that user’s password. When you click “Connect Slack” or “Sign in with GitHub” somewhere, you are going through an OAuth flow. The app you are connecting to (Slack, GitHub, or anything else) is called the provider. The temporary credential the provider hands back, called an access token, is what your app then presents on every future request as proof that this specific user said yes.

That single sentence, this specific user said yes, is the part most quick agent tutorials skip. They wire up one API key, call it good, and move on. That works fine for a single-user script. It breaks the moment your agent serves more than one person, because now every tool call has to answer an identity question before it answers a task question: who is this for, right now?

You will build:

• Two small FastAPI services, ChatterHub and IssueDock, each a real (if simplified) OAuth 2.0 authorization server and API, standing in for a team chat app like Slack and an issue tracker like GitHub.
• An encrypted, per-user, per-provider token store backed by SQLite and the cryptography library’s Fernet symmetric encryption.
• An OAuth consent flow: a user visits a connect link, approves access, and your app exchanges the resulting code for a token without any token ever touching the browser.
• A local AI agent, using a small model served by Ollama, that reads recent chat messages, decides which ones are real action items, and files issues using the token that belongs to whoever it is currently serving.
• A live demonstration that a naive, single shared token leaks one user’s data to another, followed by the fixed version proving that it cannot.

Two topics stay out of scope: the Model Context Protocol (MCP) and voice or realtime agent hosts. The identity pattern you learn here applies the same way in both, but the surrounding plumbing is a different article. If you want the OAuth 2.0 Authorization Code flow explained from first principles, including PKCE, see sxz.io’s earlier How to Build an OAuth 2.0 Authorization Code Flow With PKCE in Python. This tutorial assumes that background and focuses on the part that article does not cover: what changes when the same app has to do this for many different users, safely, inside an agent’s tool loop.

Prerequisites

• Python 3.11 or newer installed (this tutorial was tested on Python 3.13.14). Check your version with python --version.
• pip for installing packages, and basic comfort using a terminal.
• Ollama installed, to run a small model locally. No cloud API key is required or used anywhere in this tutorial.
• Basic familiarity with HTTP: what a GET and a POST request are, and what a redirect is. Prior OAuth experience helps but is not required; the mechanics are explained as you go.
• No real Slack or GitHub developer account is needed. You will build local stand-in providers so the whole tutorial is reproducible offline, and the last section explains exactly what changes when you point the same code at a real provider.

The Problem: Why a Single Shared Token Breaks a Multi-User Agent

The fastest way to wire an agent up to an external tool is to generate one API token, usually under whoever set the integration up, and hardcode it. The agent calls the tool, the tool checks the token, the token is valid, the call goes through. For a single person automating their own workflow, this is completely fine.

The trouble starts when that same agent is supposed to serve a team. The code has no way to distinguish “Alice is asking” from “Bob is asking”, because it only ever holds one credential, and that credential belongs to whoever originally set it up. Every read uses that one person’s workspace access. Every write happens under that one person’s name. If your agent’s logic ever assumes “the current user” without actually checking which token backs that assumption, you have built a tool that quietly acts as one identity no matter who is really asking.

The fix has two parts, and you will implement both:

• Each user connects their own account separately. Alice authorizes ChatterHub for herself. Bob authorizes it for himself, independently.
• The agent’s code never hardcodes a token. It resolves an identifier, such as "alice", into a token at the exact moment it needs to make a call, and that token never leaks into a log line, a model prompt, or a tool schema.

Architecture Overview

Real Slack and GitHub OAuth apps require you to register a developer application, get a client ID and secret, and point a callback URL at a public HTTPS address. None of that is available inside a disposable tutorial sandbox, and requiring readers to set up two developer accounts before they can run a single line of code is a bad way to teach the actual lesson. So this tutorial builds two small, honest stand-ins instead:

• ChatterHub, a mock team-chat provider (plays the role Slack would play), with a real OAuth 2.0 authorize/token/revoke flow and a fake but structurally real messages API.
• IssueDock, a mock issue-tracker provider (plays the role GitHub would play), with the same real OAuth 2.0 flow and an issue-creation API.

Everything about identity resolution, encrypted storage, token refresh, and cross-user isolation in this tutorial is completely real. Only the providers themselves are local stand-ins instead of the real Slack and GitHub. The last section of this tutorial shows exactly what to change to point the same code at real providers.

Five pieces make up the full system, and you will build them in this order:

• token_store.py: an encrypted, per-user, per-provider token store on top of SQLite.
• provider_chatterhub.py and provider_issuedock.py: the two mock OAuth providers, each its own small FastAPI app.
• agent_app.py: the OAuth app a user actually connects to. It starts the consent flow and receives the callback.
• agent.py: the tool-calling agent loop, which reads a user’s ChatterHub channel, asks a local model whether each message is worth acting on, and files IssueDock issues using that same user’s token.

Everything runs on 127.0.0.1 on four different ports: ChatterHub on 9101, IssueDock on 9102, the agent’s OAuth app on 9100, and Ollama’s own API on its default 11434.

Step 1: Set Up Your Environment

Install the Python Dependencies

Create a project folder, a virtual environment, and install the four packages this tutorial uses. FastAPI needs python-multipart specifically to parse the application/x-www-form-urlencoded bodies the OAuth token endpoint receives; without it, await request.form() raises an error the moment a real request hits it.

mkdir oauth-agent-tutorial && cd oauth-agent-tutorial
python -m venv venv
source venv/bin/activate   # on Windows: venv\Scripts\activate
pip install fastapi "uvicorn[standard]" cryptography requests python-multipart

This tutorial was tested with fastapi==0.141.1, uvicorn==0.52.1, cryptography==50.0.0, requests==2.34.2, and python-multipart==0.0.32. Any reasonably recent version of each should behave the same way.

Install Ollama and Pull a Model

Ollama is a small program that downloads open-weight language models and serves them over a local HTTP API, so you can run a real model without sending anything to a cloud provider or holding an API key. Install it for your platform from ollama.com, then start the server (on most installs it is already running as a background service; if not, run ollama serve in its own terminal) and pull a model:

ollama pull qwen2.5:1.5b

This tutorial started with the much smaller qwen2.5:0.5b (397 MB) and switched to qwen2.5:1.5b (986 MB) after the smaller model turned out to be unreliable at the specific classification task this agent needs. That failure, and exactly what fixed it, is covered in detail in Step 7, because it is a real lesson worth seeing rather than skipping past.

Confirm the server is up:

curl http://127.0.0.1:11434/api/version

{"version":"0.32.9"}

Step 2: Build an Encrypted, Per-User Token Store

Before writing any OAuth code, build the piece everything else depends on: somewhere to keep tokens that is keyed correctly and never stores anything in plain text. If an attacker, or just a careless log statement, ever gets read access to this database, encrypted tokens they cannot use are far better than a table of live bearer credentials.

Create token_store.py:

import os
import sqlite3
import time

from cryptography.fernet import Fernet

DB_PATH = "tokens.db"
KEY_PATH = "token_store.key"


class NotConnectedError(Exception):
    pass


def _load_key():
    if not os.path.exists(KEY_PATH):
        key = Fernet.generate_key()
        with open(KEY_PATH, "wb") as f:
            f.write(key)
    with open(KEY_PATH, "rb") as f:
        return f.read()


_fernet = Fernet(_load_key())


def _conn():
    conn = sqlite3.connect(DB_PATH)
    conn.execute(
        """
        CREATE TABLE IF NOT EXISTS tokens (
            user_id TEXT NOT NULL,
            provider TEXT NOT NULL,
            access_token_enc BLOB NOT NULL,
            refresh_token_enc BLOB,
            expires_at REAL NOT NULL,
            PRIMARY KEY (user_id, provider)
        )
        """
    )
    return conn


def save_token(user_id, provider, access_token, refresh_token, expires_in):
    expires_at = time.time() + expires_in
    access_enc = _fernet.encrypt(access_token.encode())
    refresh_enc = _fernet.encrypt(refresh_token.encode()) if refresh_token else None
    conn = _conn()
    conn.execute(
        """
        INSERT INTO tokens (user_id, provider, access_token_enc, refresh_token_enc, expires_at)
        VALUES (?, ?, ?, ?, ?)
        ON CONFLICT(user_id, provider) DO UPDATE SET
            access_token_enc = excluded.access_token_enc,
            refresh_token_enc = excluded.refresh_token_enc,
            expires_at = excluded.expires_at
        """,
        (user_id, provider, access_enc, refresh_enc, expires_at),
    )
    conn.commit()
    conn.close()


def _row(user_id, provider):
    conn = _conn()
    row = conn.execute(
        "SELECT access_token_enc, refresh_token_enc, expires_at FROM tokens "
        "WHERE user_id = ? AND provider = ?",
        (user_id, provider),
    ).fetchone()
    conn.close()
    return row


def has_connection(user_id, provider):
    return _row(user_id, provider) is not None


def get_valid_access_token(user_id, provider, refresh_fn):
    row = _row(user_id, provider)
    if row is None:
        raise NotConnectedError(f"{user_id} has not connected {provider}")

    access_enc, refresh_enc, expires_at = row

    if time.time() < expires_at - 5:
        return _fernet.decrypt(access_enc).decode()

    if refresh_enc is None:
        raise NotConnectedError(
            f"{user_id}'s {provider} grant expired and has no refresh token; reconnect required"
        )

    refresh_token = _fernet.decrypt(refresh_enc).decode()
    try:
        new_access, new_refresh, expires_in = refresh_fn(refresh_token)
    except Exception as exc:
        # A dead grant (revoked by the user, or expired on the provider's side) is not
        # a crash, it is the same "not connected" state as never having authorized at
        # all. The caller should not have to tell the two apart.
        raise NotConnectedError(
            f"{user_id}'s {provider} grant is no longer valid; reconnect required"
        ) from exc
    save_token(user_id, provider, new_access, new_refresh or refresh_token, expires_in)
    return new_access

A few details matter here:

• The table’s primary key is (user_id, provider), not just user_id. One user can, and in this tutorial does, hold separate independent grants for ChatterHub and IssueDock at the same time.
• Fernet (from the cryptography package) is symmetric authenticated encryption: the same key that encrypts a token is required to decrypt it, and any tampering with the ciphertext is detected rather than silently accepted. The key itself lives in a local file, token_store.key, generated the first time the store is used. In a real deployment this key belongs in a secrets manager, not a file next to your code; keeping it as a plain file here is a tutorial simplification, not a production pattern.
• get_valid_access_token is the only function the rest of the codebase calls to get a usable token. It raises a dedicated NotConnectedError the moment there is nothing safe to return, whether that is because the user never connected this provider at all, or their grant is no longer valid. Callers get one clear, catchable failure mode instead of having to guess which internal condition to handle.

Step 3: Build Two Mock OAuth Providers

Both providers implement the same shape: a browser-facing /oauth/authorize endpoint that approves a user and redirects back with a short-lived code, a server-to-server /oauth/token endpoint that exchanges that code (or a refresh token) for an access token, and a protected API that only responds to a valid bearer token. Real providers add far more (scopes, consent screens, rate limits), but this is the real skeleton underneath all of it.

ChatterHub: a Stand-In for a Team Chat App

Create provider_chatterhub.py. It seeds two fake workspaces, one for Alice and one for Bob, each with a handful of realistic chat messages:

import secrets
import time

from fastapi import FastAPI, HTTPException, Request
from fastapi.responses import RedirectResponse

app = FastAPI(title="ChatterHub (mock team-chat provider)")

CLIENT_ID = "chatterhub-demo-client"
CLIENT_SECRET = "chatterhub-demo-secret"
ACCESS_TTL = 20  # seconds; shortened from a real provider's TTL so the tutorial can show refresh happening

_auth_codes = {}       # code -> user_id
_access_tokens = {}    # token -> {"user_id": ..., "expires_at": ...}
_refresh_tokens = {}   # refresh_token -> user_id

WORKSPACES = {
    "alice": [
        {"id": "m1", "text": "Can someone restart the payments worker, it has been down since 9am"},
        {"id": "m2", "text": "lol nice coffee run today"},
        {"id": "m3", "text": "The nightly export job is failing with a KeyError on region, someone should look at it"},
    ],
    "bob": [
        {"id": "m1", "text": "reminder: standup moved to 10:15 tomorrow"},
        {"id": "m2", "text": "Login page throws a 500 for users with no avatar set, needs a fix"},
    ],
}


@app.get("/oauth/authorize")
def authorize(client_id: str, redirect_uri: str, state: str, user: str):
    """Stands in for the page a real provider shows where a user reviews and approves access."""
    if client_id != CLIENT_ID:
        raise HTTPException(400, "unknown client_id")
    if user not in WORKSPACES:
        raise HTTPException(404, f"no such ChatterHub user: {user}")
    code = secrets.token_urlsafe(24)
    _auth_codes[code] = user
    return RedirectResponse(f"{redirect_uri}?code={code}&state={state}&provider=chatterhub")


@app.post("/oauth/token")
async def token(request: Request):
    form = await request.form()
    if form.get("client_id") != CLIENT_ID or form.get("client_secret") != CLIENT_SECRET:
        raise HTTPException(401, "invalid client credentials")

    grant_type = form.get("grant_type")
    if grant_type == "authorization_code":
        code = form.get("code")
        user_id = _auth_codes.pop(code, None)
        if not user_id:
            raise HTTPException(400, "invalid or already-used code")
    elif grant_type == "refresh_token":
        rt = form.get("refresh_token")
        user_id = _refresh_tokens.get(rt)
        if not user_id:
            raise HTTPException(400, "invalid refresh_token")
    else:
        raise HTTPException(400, "unsupported grant_type")

    access_token = "cht_" + secrets.token_urlsafe(24)
    refresh_token = "chr_" + secrets.token_urlsafe(24)
    _access_tokens[access_token] = {"user_id": user_id, "expires_at": time.time() + ACCESS_TTL}
    _refresh_tokens[refresh_token] = user_id
    return {
        "access_token": access_token,
        "refresh_token": refresh_token,
        "expires_in": ACCESS_TTL,
        "token_type": "bearer",
    }


def _authed_user(auth_header):
    if not auth_header or not auth_header.startswith("Bearer "):
        raise HTTPException(401, "missing bearer token")
    tok = auth_header.removeprefix("Bearer ")
    entry = _access_tokens.get(tok)
    if not entry or entry["expires_at"] < time.time():
        raise HTTPException(401, "token invalid or expired")
    return entry["user_id"]


@app.get("/api/channels/{channel}/messages")
def list_messages(channel: str, request: Request):
    user_id = _authed_user(request.headers.get("authorization"))
    return {"user": user_id, "channel": channel, "messages": WORKSPACES[user_id]}


@app.post("/api/channels/{channel}/reply")
async def reply(channel: str, request: Request):
    user_id = _authed_user(request.headers.get("authorization"))
    body = await request.json()
    return {"ok": True, "user": user_id, "channel": channel, "posted": body.get("text")}


@app.post("/oauth/revoke")
async def revoke(request: Request):
    """Mirrors RFC 7009: a user disconnecting the app on the provider's side. The
    refresh token stops working immediately; any access token issued from it keeps
    working only until it naturally expires."""
    form = await request.form()
    rt = form.get("token")
    _refresh_tokens.pop(rt, None)
    return {"revoked": True}

Notice the access token time-to-live, ACCESS_TTL = 20 seconds. A real provider’s tokens typically last much longer (GitHub’s classic OAuth tokens do not expire by default at all, and Slack’s rotate only if the app opts into token rotation). Twenty seconds is deliberately short so that later in this tutorial you can watch a real token expire and get refreshed within a normal test run, instead of waiting an hour.

The /oauth/revoke endpoint mirrors RFC 7009, the OAuth Token Revocation standard: a user disconnecting the app on the provider’s side. You will use it in Step 9 to prove that a dead grant fails safely instead of crashing.

IssueDock: a Stand-In for an Issue Tracker

Create provider_issuedock.py. The structure is deliberately almost identical to ChatterHub’s, because in a real OAuth 2.0 integration it usually is: the protocol does not change from provider to provider, only the API underneath it does.

import secrets
import time

from fastapi import FastAPI, HTTPException, Request
from fastapi.responses import RedirectResponse

app = FastAPI(title="IssueDock (mock issue-tracker provider)")

CLIENT_ID = "issuedock-demo-client"
CLIENT_SECRET = "issuedock-demo-secret"
ACCESS_TTL = 20

_auth_codes = {}
_access_tokens = {}
_refresh_tokens = {}
_issues = {}  # user_id -> list of issue dicts
_known_users = {"alice", "bob"}


@app.get("/oauth/authorize")
def authorize(client_id: str, redirect_uri: str, state: str, user: str):
    if client_id != CLIENT_ID:
        raise HTTPException(400, "unknown client_id")
    if user not in _known_users:
        raise HTTPException(404, f"no such IssueDock user: {user}")
    code = secrets.token_urlsafe(24)
    _auth_codes[code] = user
    return RedirectResponse(f"{redirect_uri}?code={code}&state={state}&provider=issuedock")


@app.post("/oauth/token")
async def token(request: Request):
    form = await request.form()
    if form.get("client_id") != CLIENT_ID or form.get("client_secret") != CLIENT_SECRET:
        raise HTTPException(401, "invalid client credentials")

    grant_type = form.get("grant_type")
    if grant_type == "authorization_code":
        code = form.get("code")
        user_id = _auth_codes.pop(code, None)
        if not user_id:
            raise HTTPException(400, "invalid or already-used code")
    elif grant_type == "refresh_token":
        rt = form.get("refresh_token")
        user_id = _refresh_tokens.get(rt)
        if not user_id:
            raise HTTPException(400, "invalid refresh_token")
    else:
        raise HTTPException(400, "unsupported grant_type")

    access_token = "idt_" + secrets.token_urlsafe(24)
    refresh_token = "idr_" + secrets.token_urlsafe(24)
    _access_tokens[access_token] = {"user_id": user_id, "expires_at": time.time() + ACCESS_TTL}
    _refresh_tokens[refresh_token] = user_id
    return {
        "access_token": access_token,
        "refresh_token": refresh_token,
        "expires_in": ACCESS_TTL,
        "token_type": "bearer",
    }


def _authed_user(auth_header):
    if not auth_header or not auth_header.startswith("Bearer "):
        raise HTTPException(401, "missing bearer token")
    tok = auth_header.removeprefix("Bearer ")
    entry = _access_tokens.get(tok)
    if not entry or entry["expires_at"] < time.time():
        raise HTTPException(401, "token invalid or expired")
    return entry["user_id"]


@app.post("/api/repos/{repo}/issues")
async def create_issue(repo: str, request: Request):
    user_id = _authed_user(request.headers.get("authorization"))
    body = await request.json()
    issue_id = len(_issues.get(user_id, [])) + 1
    issue = {
        "id": issue_id,
        "repo": repo,
        "title": body["title"],
        "body": body.get("body", ""),
        "author": user_id,
    }
    _issues.setdefault(user_id, []).append(issue)
    return issue


@app.get("/api/repos/{repo}/issues")
def list_issues(repo: str, request: Request):
    user_id = _authed_user(request.headers.get("authorization"))
    return {"user": user_id, "issues": [i for i in _issues.get(user_id, []) if i["repo"] == repo]}

Step 4: Build the Agent’s Own OAuth App

This is the piece that plays the role of your actual application, the one a user grants access to. Create agent_app.py:

import secrets

import requests
from fastapi import FastAPI, HTTPException
from fastapi.responses import RedirectResponse

import token_store

app = FastAPI(title="channel-watcher-agent (OAuth app)")

PROVIDERS = {
    "chatterhub": {
        "authorize_url": "http://127.0.0.1:9101/oauth/authorize",
        "token_url": "http://127.0.0.1:9101/oauth/token",
        "client_id": "chatterhub-demo-client",
        "client_secret": "chatterhub-demo-secret",
    },
    "issuedock": {
        "authorize_url": "http://127.0.0.1:9102/oauth/authorize",
        "token_url": "http://127.0.0.1:9102/oauth/token",
        "client_id": "issuedock-demo-client",
        "client_secret": "issuedock-demo-secret",
    },
}

REDIRECT_URI = "http://127.0.0.1:9100/oauth/callback"
_pending_state = {}  # state -> user_id issuing the connect request (CSRF check on callback)


@app.get("/connect/{provider}")
def connect(provider: str, user: str):
    """A user's browser lands here when they click 'Connect ChatterHub' or 'Connect IssueDock'."""
    if provider not in PROVIDERS:
        raise HTTPException(404, "unknown provider")
    cfg = PROVIDERS[provider]
    state = secrets.token_urlsafe(16)
    _pending_state[state] = user
    url = (
        f"{cfg['authorize_url']}?client_id={cfg['client_id']}"
        f"&redirect_uri={REDIRECT_URI}&state={state}&user={user}"
    )
    return RedirectResponse(url)


@app.get("/oauth/callback")
def callback(code: str, state: str, provider: str):
    expected_user = _pending_state.pop(state, None)
    if expected_user is None:
        raise HTTPException(400, "unknown or already-used state, possible CSRF")

    cfg = PROVIDERS[provider]
    resp = requests.post(
        cfg["token_url"],
        data={
            "grant_type": "authorization_code",
            "code": code,
            "client_id": cfg["client_id"],
            "client_secret": cfg["client_secret"],
            "redirect_uri": REDIRECT_URI,
        },
    )
    resp.raise_for_status()
    payload = resp.json()

    token_store.save_token(
        expected_user,
        provider,
        payload["access_token"],
        payload.get("refresh_token"),
        payload["expires_in"],
    )
    return {"connected": provider, "user": expected_user}

/connect/{provider} is the link a user’s browser would hit when they click “Connect ChatterHub” somewhere in a real UI. It generates a random state value and remembers which user requested it, then redirects to the provider’s authorize page. /oauth/callback is where the provider redirects back to after approval. It checks the returned state against what it remembers issuing, and rejects anything it does not recognize.

That state check is a real CSRF defense, not decoration: without it, an attacker could trick a logged-in user’s browser into completing an OAuth grant the attacker initiated, potentially binding the attacker’s own external account to the victim’s session. It is the same synchronizer-token idea covered in more depth in sxz.io’s CSRF and synchronizer tokens tutorial, applied here to the OAuth callback specifically.

One real-world detail worth knowing even though this tutorial does not need it: a real provider requires the redirect URI to be HTTPS, not plain http://127.0.0.1. Locally, that usually means a tool like mkcert to generate a trusted local certificate. This tutorial runs everything over plain HTTP on localhost to keep setup to a single pip install, which is fine for learning the identity pattern but is not how you would wire this up against Slack or GitHub for real.

Step 5: Start Everything and Connect Alice

Open three separate terminals (or run each in the background) inside your project folder, with the virtual environment active in each:

uvicorn provider_chatterhub:app --host 127.0.0.1 --port 9101
uvicorn provider_issuedock:app --host 127.0.0.1 --port 9102
uvicorn agent_app:app --host 127.0.0.1 --port 9100

Now create a tiny helper, connect.py, that simulates what a user’s browser does when they click a connect link and walk through consent. The Python requests library follows redirects by default, so a single GET request here walks the full chain: agent app to provider’s authorize page to provider’s redirect back to the agent app’s callback, exactly the same hops a real browser would take.

import sys

import requests

AGENT_APP = "http://127.0.0.1:9100"


def connect(provider, user):
    """Stands in for a user's browser clicking 'Connect' and walking through consent.
    requests follows both redirects (agent app -> provider -> agent app callback) the
    same way a browser would."""
    resp = requests.get(f"{AGENT_APP}/connect/{provider}", params={"user": user})
    resp.raise_for_status()
    return resp.json()


if __name__ == "__main__":
    provider, user = sys.argv[1], sys.argv[2]
    result = connect(provider, user)
    print(result)

Connect Alice to ChatterHub:

python -c "import connect; print(connect.connect('chatterhub', 'alice'))"

{'connected': 'chatterhub', 'user': 'alice'}

Behind that one line, a real code-for-token exchange just happened: ChatterHub issued a short-lived authorization code, the agent app’s callback traded that code for an access token and a refresh token over a direct server-to-server POST, and token_store.py encrypted both before writing them to tokens.db. You can prove that last part yourself:

python -c "
import sqlite3
conn = sqlite3.connect('tokens.db')
for row in conn.execute('SELECT user_id, provider, access_token_enc FROM tokens'):
    print(row[0], row[1], row[2][:40])
"

alice chatterhub b'gAAAAABqfM7VpRF90qvFS-ncLlJxKy0nNkUEYnYq'

That gAAAAA... prefix is Fernet’s own versioned token format. There is no way to recover the real ChatterHub access token from that blob without the key in token_store.key, which is exactly the point.

Step 6: Watch the Naive Shared-Token Pattern Fail, Live

Before building the fix, it is worth seeing the failure with your own eyes rather than taking it on faith. Create naive_shared_demo.py:

import requests

CHATTERHUB_API = "http://127.0.0.1:9101"


def naive_run_for_user(user_id_the_agent_thinks_it_is_serving, shared_token):
    """The anti-pattern this tutorial fixes: one token, configured once, reused for
    every user's request no matter who the agent believes it is acting for."""
    resp = requests.get(
        f"{CHATTERHUB_API}/api/channels/engineering/messages",
        headers={"Authorization": f"Bearer {shared_token}"},
    )
    resp.raise_for_status()
    data = resp.json()
    print(f"  agent believes it is serving: {user_id_the_agent_thinks_it_is_serving!r}")
    print(f"  ChatterHub says this token actually belongs to: {data['user']!r}")
    print(f"  messages returned belong to: {data['user']!r} (private channel content)")
    for m in data["messages"]:
        print(f"    - {m['text']}")
    return data

This simulates the exact anti-pattern described in Step 6: a token obtained once, then reused for every user’s request regardless of who the agent believes it is serving. Run it using the real token Alice’s connection produced a moment ago:

python -c "
import token_store, agent, naive_shared_demo
shared_token = token_store.get_valid_access_token('alice', 'chatterhub', agent._refresh_chatterhub)
naive_shared_demo.naive_run_for_user('bob', shared_token=shared_token)
"

  agent believes it is serving: 'bob'
  ChatterHub says this token actually belongs to: 'alice'
  messages returned belong to: 'alice' (private channel content)
    - Can someone restart the payments worker, it has been down since 9am
    - lol nice coffee run today
    - The nightly export job is failing with a KeyError on region, someone should look at it

The code believes it is handling a request for Bob. ChatterHub, correctly, has no idea who the code thinks it is serving; it only knows which real grant the token belongs to, and hands back Alice’s private channel content. This is the entire bug in one live request: a shared credential carries no identity of its own, so nothing downstream can catch the mismatch. The fix is not a validation check you bolt on afterward. It is resolving a token per user, every time, which is exactly what the rest of this tutorial builds.

Step 7: Build the Tool-Calling Agent Loop

Classifying Messages With a Local Model

The agent’s job on each run is simple to state: read a channel, decide which messages describe real, actionable work, and file an issue for those. The “decide” part is where a language model earns its place, versus a plain keyword filter that would miss “the nightly export job is failing” just as easily as it would misfire on “lol nice coffee run today”.

The first version of this tutorial’s classifier asked the model to reply with one line of free text in a fixed format, ACTION: yes | TITLE: ..., and parsed that with a regular expression. It did not survive contact with a real model. Here is what actually happened running it against qwen2.5:0.5b, copied directly from that session:

  [alice] msg='m1' model_output='YES | Payments Worker restarted - This is a known issue that should be tracked as a bug.' -> action=False
  [alice] msg='m2' model_output='ACTION: yes | TITLE: none' -> action=True
  [alice] msg='m3' model_output='YES | Security Bug' -> action=False

Two separate problems are visible in those three lines. First, a parsing problem: the model answered “YES” instead of the exact string “ACTION: yes” twice, so the regular expression simply did not match and both real bugs (a downed payments worker, a failing export job) were silently treated as not action items. Second, a model-quality problem: even where the format matched, the model classified “lol nice coffee run today” as an action item with the literal title “none”. A small model can follow instructions inconsistently, and free-text parsing has no way to tell a formatting failure apart from a real “no”.

The fix has two independent parts, and both matter on their own.

The first is to stop parsing free text at all. Ollama supports schema-constrained structured output: pass a JSON schema in the request’s format field, and the model is constrained to return only JSON matching that shape. This does not make the model’s judgement better, but it makes its output mechanically reliable to read, which is a real and separate improvement:

curl http://127.0.0.1:11434/api/generate -d '{
  "model": "qwen2.5:0.5b",
  "prompt": "Message: lol nice coffee run today",
  "stream": false,
  "format": {
    "type": "object",
    "properties": {
      "is_action_item": {"type": "boolean"},
      "title": {"type": "string"}
    },
    "required": ["is_action_item", "title"]
  }
}'

The second is that structured output alone was not enough. With the schema in place but still on the 0.5B model, the classifications themselves stayed unreliable, in one test run every single message came back false, including two genuine bugs. Moving up to qwen2.5:1.5b (986 MB, still small and fast enough to run on a CPU) and adding four few-shot examples directly in the prompt fixed it completely: five out of five test messages, including all three tricky ones from the ChatterHub seed data, classified correctly. The lesson generalizes past this one tutorial: structured output fixes whether you can parse a model’s answer, not whether the answer is right, and a model that is too small for a judgement task will stay unreliable no matter how you format its output.

Create agent.py with the version that actually works:

import json

import requests

import token_store

OLLAMA_URL = "http://127.0.0.1:11434/api/generate"
MODEL = "qwen2.5:1.5b"

CLASSIFY_SCHEMA = {
    "type": "object",
    "properties": {
        "is_action_item": {"type": "boolean"},
        "title": {"type": "string"},
    },
    "required": ["is_action_item", "title"],
}

FEWSHOT_PROMPT = """You triage team chat messages for a software team. Decide if a message \
describes a bug or a concrete action item that should become a tracked issue. Small talk, \
jokes, and scheduling reminders are NOT action items, only real technical problems or \
explicit requests for someone to do something are.

Examples:
Message: "lol that deploy meme is so accurate"
{"is_action_item": false, "title": ""}

Message: "the checkout API is returning 500s for guest users since the last deploy"
{"is_action_item": true, "title": "Checkout API returns 500 for guest users"}

Message: "anyone up for lunch at noon?"
{"is_action_item": false, "title": ""}

Message: "can someone rotate the expired TLS cert on api.internal, it's blocking every request"
{"is_action_item": true, "title": "Rotate expired TLS cert on api.internal"}

Now classify this message the same way:
Message: "%s"
"""

CHATTERHUB_API = "http://127.0.0.1:9101"
ISSUEDOCK_API = "http://127.0.0.1:9102"


def _refresh_chatterhub(refresh_token):
    r = requests.post(
        "http://127.0.0.1:9101/oauth/token",
        data={
            "grant_type": "refresh_token",
            "refresh_token": refresh_token,
            "client_id": "chatterhub-demo-client",
            "client_secret": "chatterhub-demo-secret",
        },
    )
    r.raise_for_status()
    p = r.json()
    return p["access_token"], p.get("refresh_token"), p["expires_in"]


def _refresh_issuedock(refresh_token):
    r = requests.post(
        "http://127.0.0.1:9102/oauth/token",
        data={
            "grant_type": "refresh_token",
            "refresh_token": refresh_token,
            "client_id": "issuedock-demo-client",
            "client_secret": "issuedock-demo-secret",
        },
    )
    r.raise_for_status()
    p = r.json()
    return p["access_token"], p.get("refresh_token"), p["expires_in"]


def classify_message(text):
    """Ask the local model whether a chat message describes real, actionable work.

    Uses Ollama's schema-constrained structured output (the `format` field) instead of
    asking for free text and regex-parsing it. The model is still free to get the
    *classification* wrong, but it can no longer return a reply the parser cannot read."""
    r = requests.post(
        OLLAMA_URL,
        json={
            "model": MODEL,
            "prompt": FEWSHOT_PROMPT % text,
            "stream": False,
            "options": {"temperature": 0},
            "format": CLASSIFY_SCHEMA,
        },
        timeout=60,
    )
    r.raise_for_status()
    raw = r.json()["response"].strip()
    parsed = json.loads(raw)
    is_action = bool(parsed["is_action_item"])
    title = parsed["title"].strip() if is_action else None
    return is_action, title, raw


def run_for_user(user_id, channel="engineering"):
    """Read a user's ChatterHub channel, file IssueDock issues for real action items,
    and reply in-thread, all scoped to that user's own tokens."""
    chat_token = token_store.get_valid_access_token(user_id, "chatterhub", _refresh_chatterhub)

    resp = requests.get(
        f"{CHATTERHUB_API}/api/channels/{channel}/messages",
        headers={"Authorization": f"Bearer {chat_token}"},
    )
    resp.raise_for_status()
    messages = resp.json()["messages"]

    filed = []
    for msg in messages:
        is_action, title, raw = classify_message(msg["text"])
        print(f"  [{user_id}] msg={msg['id']!r} model_output={raw!r} -> action={is_action}")
        if not is_action:
            continue

        # Resolved late: we only look up the IssueDock token for this user at the
        # moment we actually need to make the call, not earlier.
        issue_token = token_store.get_valid_access_token(user_id, "issuedock", _refresh_issuedock)
        issue_resp = requests.post(
            f"{ISSUEDOCK_API}/api/repos/team-repo/issues",
            headers={"Authorization": f"Bearer {issue_token}"},
            json={"title": title, "body": f"Filed by channel-watcher-agent from message: {msg['text']}"},
        )
        issue_resp.raise_for_status()
        issue = issue_resp.json()

        chat_token = token_store.get_valid_access_token(user_id, "chatterhub", _refresh_chatterhub)
        requests.post(
            f"{CHATTERHUB_API}/api/channels/{channel}/reply",
            headers={"Authorization": f"Bearer {chat_token}"},
            json={"text": f"Filed issue #{issue['id']}: {issue['title']}"},
        )
        filed.append(issue)

    return filed

Resolving Tokens Late, Never Logging Them

Look at the shape of run_for_user closely. It does not resolve every token it might need up front. It fetches the ChatterHub token, uses it immediately, and only resolves an IssueDock token inside the loop, at the exact point where an issue is about to be filed, and only for messages the classifier actually flagged. That is deliberate: the longer a decrypted token sits in a variable, the more places it could accidentally end up (a log statement, an exception traceback, a debugger session). Resolving late shrinks that window to as close to zero as the code allows. The print statement inside the loop logs the model’s classification output, never the token itself; that separation is not an accident either.

Step 8: Run the Agent, Per User, and Prove Isolation

Connect Alice to IssueDock as well, and connect Bob to ChatterHub only, deliberately leaving his IssueDock connection missing:

python -c "import connect; print(connect.connect('issuedock', 'alice'))"
python -c "import connect; print(connect.connect('chatterhub', 'bob'))"

{'connected': 'issuedock', 'user': 'alice'}
{'connected': 'chatterhub', 'user': 'bob'}

Run the agent for Alice, who is fully connected to both providers:

python -c "import agent; print(agent.run_for_user('alice'))"

  [alice] msg='m1' model_output='{"is_action_item": true, "title": "Restart Payments Worker"}' -> action=True
  [alice] msg='m2' model_output='{"is_action_item": false, "title": ""}' -> action=False
  [alice] msg='m3' model_output='{"is_action_item": true, "title": "Nightly export job fails with KeyError in region"}' -> action=True

[{'id': 1, 'repo': 'team-repo', 'title': 'Restart Payments Worker', 'body': 'Filed by channel-watcher-agent from message: Can someone restart the payments worker, it has been down since 9am', 'author': 'alice'}, {'id': 2, 'repo': 'team-repo', 'title': 'Nightly export job fails with KeyError in region', 'body': 'Filed by channel-watcher-agent from message: The nightly export job is failing with a KeyError on region, someone should look at it', 'author': 'alice'}]

Two real issues, filed under Alice’s own IssueDock grant, for the two messages that genuinely described work. The coffee message correctly produced no issue at all. Now run the agent for Bob, who has not connected IssueDock:

python -c "
import agent, token_store
try:
    print(agent.run_for_user('bob'))
except token_store.NotConnectedError as e:
    print(f'blocked safely: {e}')
"

  [bob] msg='m1' model_output='{"is_action_item": false, "title": ""}' -> action=False
  [bob] msg='m2' model_output='{"is_action_item": true, "title": "Fix login page error for users without avatars"}' -> action=True
blocked safely: bob has not connected issuedock

Bob’s standup reminder is correctly ignored, and his real login bug is correctly classified as an action item. Then, at the exact moment the code needs to file it, token_store.get_valid_access_token raises NotConnectedError instead of the request going through. This is the isolation guarantee working as designed: there was no fallback to Alice’s IssueDock token, no silent no-op, just a clean, catchable error that tells you precisely what is missing.

Verify Isolation Independently of the Agent’s Word

Do not just trust the agent’s own error message. Check the underlying state directly:

python -c "
import requests, token_store, agent
alice_token = token_store.get_valid_access_token('alice', 'issuedock', agent._refresh_issuedock)
r = requests.get('http://127.0.0.1:9102/api/repos/team-repo/issues', headers={'Authorization': f'Bearer {alice_token}'})
print('issues visible with alice\'s own token:', r.json())
print('bob has an issuedock connection at all?', token_store.has_connection('bob', 'issuedock'))
"

issues visible with alice's own token: {'user': 'alice', 'issues': [{'id': 1, ...}, {'id': 2, ...}]}
bob has an issuedock connection at all? False

Alice’s own token, queried directly against IssueDock, returns exactly the two issues her agent run filed. Bob has no row in the token store for IssueDock at all, not an empty or invalid one, none. There is no token for the code to misuse even if a future bug tried.

Step 9: Handle Token Refresh and Revocation

Transparent Refresh Past Expiry

ChatterHub’s access tokens expire after 20 seconds in this tutorial. Watch get_valid_access_token refresh one automatically, with no reconnect required:

python -c "
import time, token_store, agent
old = token_store.get_valid_access_token('alice', 'chatterhub', agent._refresh_chatterhub)
print('before:', old[:16])
time.sleep(22)
new = token_store.get_valid_access_token('alice', 'chatterhub', agent._refresh_chatterhub)
print('after:', new[:16])
print('changed:', old != new)
"

before: cht_T0RfnPYz88bw
after: cht_-aW9J7yVbsPA
changed: True

The token genuinely changed, using the refresh token that was saved alongside the original access token, with no user interaction and no reconnect. GitHub’s classic OAuth tokens do not expire this way by default; Slack’s rotate only if the app opts into token rotation. Whichever policy a real provider uses, this same refresh path is what a production integration relies on.

A Revoked Grant Should Fail Like “Never Connected”

A refresh token can also die for reasons that have nothing to do with expiry: a user disconnects the app from the provider’s settings page, an admin revokes access, or the provider itself invalidates a grant. The first version of get_valid_access_token in this tutorial did not handle that case: it let whatever exception the refresh call raised propagate straight up as a crash. Create test_revocation.py to reproduce a dead grant using ChatterHub’s /oauth/revoke endpoint from Step 3, standing in for a user clicking disconnect:

import sqlite3
import time

import requests
from cryptography.fernet import Fernet

import agent
import token_store

SEP = "=" * 70
print(f"\n{SEP}\nSTEP G: a dead grant (revoked token) should fail the same clean way\n{SEP}")

# We do not have a real ChatterHub settings page to click "disconnect" on, so we look
# up alice's own refresh token the same way the token store itself would (using the
# same encryption key) purely to simulate her revoking access on ChatterHub's side.
# A real app never exposes a user's own refresh token back out like this.
key = open("token_store.key", "rb").read()
fernet = Fernet(key)
conn = sqlite3.connect("tokens.db")
row = conn.execute(
    "SELECT refresh_token_enc FROM tokens WHERE user_id = ? AND provider = ?",
    ("alice", "chatterhub"),
).fetchone()
real_refresh_token = fernet.decrypt(row[0]).decode()

# Force the cached access token to look expired so the next call must refresh.
conn.execute(
    "UPDATE tokens SET expires_at = ? WHERE user_id = ? AND provider = ?",
    (time.time() - 100, "alice", "chatterhub"),
)
conn.commit()
conn.close()

revoke_resp = requests.post("http://127.0.0.1:9101/oauth/revoke", data={"token": real_refresh_token})
print(f"simulated 'alice disconnects the app' on ChatterHub: {revoke_resp.json()}")

try:
    token_store.get_valid_access_token("alice", "chatterhub", agent._refresh_chatterhub)
    print("no exception raised, this would be a bug")
except token_store.NotConnectedError as e:
    print(f"caught cleanly as NotConnectedError: {e}")

Run it:

python test_revocation.py

simulated 'alice disconnects the app' on ChatterHub: {'revoked': True}
caught cleanly as NotConnectedError: alice's chatterhub grant is no longer valid; reconnect required

That clean NotConnectedError is not automatic: it comes directly from the try/except block already shown in token_store.py‘s get_valid_access_token: any exception the refresh function raises gets caught and re-raised as the same NotConnectedError a caller already knows how to handle. A dead grant and a grant that was never created look identical to every caller in this codebase, which is exactly what you want. The alternative, letting a raw HTTP error escape from deep inside a token refresh, would force every single call site to separately guess what kind of failure just happened.

Common Mistakes and Gotchas

• A single shared token for a multi-user agent. Step 6 showed this leaking one user’s private data to a request meant for someone else, live. If your agent code has exactly one hardcoded credential anywhere, it cannot tell users apart, no matter how correct the rest of the logic is.
• Regex-parsing free-text model output. The first version of this tutorial’s classifier asked for a fixed text format and parsed it with a regular expression. A real small model deviated from that format on two of three test messages, silently turning real bugs into false negatives. Schema-constrained structured output (Ollama’s format field) turns a parsing failure into a schema validation failure you cannot silently ignore.
• Assuming structured output fixes classification accuracy. It does not; it only fixes whether you can read the answer. This tutorial’s 0.5B model produced perfectly valid JSON that was still wrong on every single test message. Moving to a 1.5B model with a few-shot prompt is what actually fixed the judgement, not the schema.
• Resolving or logging a token before you need it. Resolve a user’s token immediately before the call that uses it, as run_for_user does for IssueDock, and never let a decrypted token reach a log statement, a prompt sent to a model, or a tool schema. The agent’s own print statement in this tutorial logs the classification decision, never the token.
• Treating a dead grant as a crash instead of a normal state. A revoked or otherwise invalidated refresh token is not exceptional; it is a routine state your code will hit in production. Catch it and surface the same clean “not connected” signal you would show a user who never connected in the first place, as shown in Step 9.
• Forgetting that a rotated encryption key needs every process restarted. While building this tutorial, deleting token_store.key to start fresh while the OAuth callback server was still running from an earlier session caused a real cryptography.fernet.InvalidToken error: the already-running server process had the old key cached in memory from when it first started, wrote new tokens encrypted with that old key, while a separate freshly started script generated and used a new key to try to read them. If you ever rotate or regenerate the key file, restart every process that touches the token store, not just the one you are actively working in.
• Assuming plain HTTP localhost callbacks will work against a real provider. Real OAuth providers require an HTTPS redirect URI. This tutorial uses plain HTTP against local mock providers to keep the setup to one pip install; a real integration needs a tool like mkcert for a trusted local certificate, or a real deployed HTTPS endpoint.

Applying This to a Real Provider Like Slack or GitHub

Nothing about the identity architecture changes when you swap a mock provider for a real one. What changes is entirely contained in the PROVIDERS dictionary in agent_app.py and the two small _refresh_* functions in agent.py:

• Register a real OAuth app in Slack’s or GitHub’s developer console to get a real client_id and client_secret, and set your callback’s real HTTPS URL as the registered redirect URI.
• Replace authorize_url and token_url with the provider’s real endpoints (for example, GitHub’s are https://github.com/login/oauth/authorize and https://github.com/login/oauth/access_token).
• Serve your callback over real HTTPS, using mkcert locally or a real TLS certificate in production.
• Add whatever scope parameter the provider requires so the token you get back actually has permission to read the channel or repository you need.

token_store.py, the consent-flow shape in agent_app.py, and the resolve-late pattern in agent.py do not need to change at all. That separation, protocol-shaped code that does not care which specific provider it is talking to, is the actual payoff of building this from scratch once instead of wiring up one provider’s SDK and hoping the pattern generalizes.

How to Confirm It All Works End to End

Before considering this done, confirm every piece independently:

• All three servers respond: curl http://127.0.0.1:9101/docs, :9102/docs, and :9100/docs each return HTTP 200.
• Ollama is reachable and the model is pulled: curl http://127.0.0.1:11434/api/version and ollama list shows qwen2.5:1.5b.
• Connecting a user actually stores an encrypted row: query tokens.db directly and confirm the stored bytes start with Fernet’s gAAAAA prefix, not a readable token.
• The naive shared-token demo shows the leak: running naive_shared_demo.naive_run_for_user('bob', shared_token=alice_token) returns data explicitly labeled as belonging to Alice.
• The fixed agent correctly isolates users: Alice’s run files real issues; Bob’s run raises NotConnectedError rather than silently using Alice’s IssueDock token.
• Refresh and revocation both fail or succeed the way Step 9 demonstrated, not with an unhandled exception.

Next Steps

From here, a few directions are worth exploring on your own:

• Add PKCE (Proof Key for Code Exchange) to the consent flow you built here, following the mechanics in sxz.io’s OAuth 2.0 Authorization Code flow with PKCE tutorial, for an extra layer of protection against a stolen authorization code.
• If your agent already needs to isolate more than tokens, such as separate conversation memory per user, see sxz.io’s multi-user AI agent with FastAPI and Streamlit tutorial for a complementary pattern.
• Swap the local token_store.key file for a real secrets manager (such as a cloud provider’s key vault, or HashiCorp Vault) before running anything like this in production.
• Once you are comfortable with the mock providers, try pointing the same code at a real Slack or GitHub OAuth app in a throwaway workspace and repository, using the mapping described earlier in this tutorial.