Building an MCP server with OAuth for Claude

import os

# The URLs people and clients actually reach. Set these to your HTTPS addresses.
AUTH_URL = os.environ.get("AUTH_URL", "http://localhost:9000")
MCP_URL = os.environ.get("MCP_URL", "http://localhost:8000")
RESOURCE = f"{MCP_URL}/mcp"
JWKS_URL = f"{AUTH_URL}/jwks.json"
SCOPES = ["notes:read", "notes:write"]
DATABASE = os.environ.get("DATABASE", "notes.db")

import secrets
import sqlite3

from config import DATABASE

db = sqlite3.connect(DATABASE, check_same_thread=False)
db.execute("CREATE TABLE IF NOT EXISTS notes (id TEXT PRIMARY KEY, owner TEXT, text TEXT)")


def list_for(owner: str) -> list[dict]:
    rows = db.execute("SELECT id, text FROM notes WHERE owner = ?", (owner,))
    return [{"id": row[0], "text": row[1]} for row in rows]


def add(owner: str, text: str) -> str:
    note_id = secrets.token_urlsafe(8)
    db.execute("INSERT INTO notes VALUES (?, ?, ?)", (note_id, owner, text))
    db.commit()
    return note_id


def delete(owner: str, note_id: str) -> bool:
    # owner is part of the query, so a stolen or guessed id isn't enough.
    cursor = db.execute("DELETE FROM notes WHERE id = ? AND owner = ?", (note_id, owner))
    db.commit()
    return cursor.rowcount == 1

import asyncio
from urllib.parse import urlparse

import jwt
from mcp.server import MCPServer
from mcp.server.auth.middleware.auth_context import get_access_token
from mcp.server.auth.provider import AccessToken, TokenVerifier
from mcp.server.auth.settings import AuthSettings
from mcp.server.mcpserver.exceptions import ToolError
from mcp.server.transport_security import TransportSecuritySettings
from mcp.types import ToolAnnotations

import notes
from config import AUTH_URL, JWKS_URL, MCP_URL, RESOURCE


class JwtVerifier(TokenVerifier):
    """Accept only signed, unexpired tokens issued for this exact server."""

    def __init__(self):
        self.keys = jwt.PyJWKClient(JWKS_URL, cache_keys=True)

    async def verify_token(self, token: str) -> AccessToken | None:
        try:
            key = await asyncio.to_thread(self.keys.get_signing_key_from_jwt, token)
            claims = jwt.decode(
                token,
                key.key,
                algorithms=["RS256"],  # Never read the algorithm from the token.
                audience=RESOURCE,
                issuer=AUTH_URL,
                options={"require": ["exp", "iss", "aud", "sub"]},
            )
        except jwt.PyJWTError:
            return None
        return AccessToken(
            token=token,
            client_id=claims.get("client_id", "unknown"),
            scopes=claims.get("scope", "").split(),
            expires_at=claims["exp"],
            resource=RESOURCE,
            subject=claims["sub"],
        )


mcp = MCPServer(
    "notes",
    token_verifier=JwtVerifier(),
    auth=AuthSettings(
        issuer_url=AUTH_URL,
        resource_server_url=RESOURCE,
        validate_token_resource=True,
    ),
)


def user_with(scope: str) -> str:
    """Return the signed-in user, or stop if the token lacks this scope."""
    token = get_access_token()
    if token is None or token.subject is None:
        raise ToolError("Not authenticated.")
    if scope not in token.scopes:
        raise ToolError(f"This action needs the {scope} permission. Reconnect to grant it.")
    return token.subject


@mcp.tool(annotations=ToolAnnotations(read_only_hint=True))
def list_notes() -> list[dict]:
    """List the signed-in user's notes."""
    return notes.list_for(user_with("notes:read"))


@mcp.tool(annotations=ToolAnnotations(read_only_hint=False, destructive_hint=False))
def add_note(text: str) -> str:
    """Save a note for the signed-in user and return its id."""
    if not text.strip() or len(text) > 5000:
        raise ToolError("A note must be between 1 and 5,000 characters.")
    return notes.add(user_with("notes:write"), text)


@mcp.tool(annotations=ToolAnnotations(read_only_hint=False, destructive_hint=True))
def delete_note(note_id: str) -> str:
    """Delete one of the signed-in user's notes by id."""
    if not notes.delete(user_with("notes:write"), note_id):
        raise ToolError("No note with that id.")
    return "Deleted."


app = mcp.streamable_http_app(
    stateless_http=True,
    json_response=True,
    transport_security=TransportSecuritySettings(
        allowed_hosts=[urlparse(MCP_URL).netloc],
    ),
)

"""A small OAuth 2.1 authorization server for local development.

It shows every step of the flow. In production, use a real identity provider.
"""
import html
import json
import secrets
import time
from urllib.parse import urlparse

import jwt
from cryptography.hazmat.primitives.asymmetric import rsa
from mcp.server.auth.provider import (
    AccessToken,
    AuthorizationCode,
    AuthorizationParams,
    AuthorizeError,
    RefreshToken,
    RegistrationError,
    TokenError,
    construct_redirect_uri,
)
from mcp.server.auth.routes import create_auth_routes
from mcp.server.auth.settings import AuthSettings, ClientRegistrationOptions, RevocationOptions
from mcp.shared.auth import OAuthClientInformationFull, OAuthToken
from starlette.applications import Starlette
from starlette.requests import Request
from starlette.responses import HTMLResponse, JSONResponse, RedirectResponse, Response
from starlette.routing import Route

from config import AUTH_URL, RESOURCE, SCOPES

ACCESS_TOKEN_SECONDS = 15 * 60
CODE_SECONDS = 5 * 60
PENDING_SECONDS = 10 * 60
# Claude on the web, Desktop and mobile. Claude Code uses a loopback address.
HOSTED_CALLBACKS = {"https://claude.ai/api/mcp/auth_callback"}

private_key = rsa.generate_private_key(public_exponent=65537, key_size=2048)
key_id = secrets.token_hex(4)


def allowed_redirect(uri: str) -> bool:
    parsed = urlparse(uri)
    is_loopback = parsed.scheme == "http" and parsed.hostname in {"localhost", "127.0.0.1"}
    return is_loopback or uri in HOSTED_CALLBACKS


class Provider:
    def __init__(self):
        self.clients: dict[str, OAuthClientInformationFull] = {}
        self.pending: dict[str, dict] = {}
        self.codes: dict[str, AuthorizationCode] = {}
        self.refresh_tokens: dict[str, RefreshToken] = {}

    async def get_client(self, client_id):
        return self.clients.get(client_id)

    async def register_client(self, client_info):
        if not all(allowed_redirect(str(uri)) for uri in client_info.redirect_uris):
            raise RegistrationError("invalid_redirect_uri", "Redirect URI not allowed.")
        self.clients[client_info.client_id] = client_info

    async def authorize(self, client, params: AuthorizationParams) -> str:
        if params.resource not in (None, RESOURCE):
            raise AuthorizeError("invalid_target", "Unknown resource.")
        request_id = secrets.token_urlsafe(24)
        self.pending[request_id] = {"client": client, "params": params, "at": time.time()}
        return f"{AUTH_URL}/consent?request={request_id}"

    def issue_code(self, client, params, subject, scopes) -> str:
        code = secrets.token_urlsafe(32)
        self.codes[code] = AuthorizationCode(
            code=code,
            scopes=scopes,
            expires_at=time.time() + CODE_SECONDS,
            client_id=client.client_id,
            code_challenge=params.code_challenge,
            redirect_uri=params.redirect_uri,
            redirect_uri_provided_explicitly=params.redirect_uri_provided_explicitly,
            resource=RESOURCE,  # This token works on this server and nowhere else.
            subject=subject,
        )
        return code

    async def load_authorization_code(self, client, authorization_code):
        code = self.codes.get(authorization_code)
        return code if code and code.client_id == client.client_id else None

    def tokens(self, client_id, subject, scopes, resource) -> OAuthToken:
        now = int(time.time())
        access = jwt.encode(
            {
                "iss": AUTH_URL,
                "sub": subject,
                "aud": resource,
                "client_id": client_id,
                "scope": " ".join(scopes),
                "iat": now,
                "exp": now + ACCESS_TOKEN_SECONDS,
                "jti": secrets.token_hex(8),
            },
            private_key,
            algorithm="RS256",
            headers={"kid": key_id},
        )
        refresh = secrets.token_urlsafe(32)
        self.refresh_tokens[refresh] = RefreshToken(
            token=refresh, client_id=client_id, scopes=scopes, resource=resource, subject=subject
        )
        return OAuthToken(
            access_token=access,
            expires_in=ACCESS_TOKEN_SECONDS,
            scope=" ".join(scopes),
            refresh_token=refresh,
        )

    async def exchange_authorization_code(self, client, authorization_code):
        del self.codes[authorization_code.code]  # A code works once.
        return self.tokens(
            client.client_id,
            authorization_code.subject,
            authorization_code.scopes,
            authorization_code.resource,
        )

    async def load_refresh_token(self, client, refresh_token):
        token = self.refresh_tokens.get(refresh_token)
        return token if token and token.client_id == client.client_id else None

    async def exchange_refresh_token(self, client, refresh_token, scopes):
        del self.refresh_tokens[refresh_token.token]  # Rotate on every use.
        if not set(scopes) <= set(refresh_token.scopes):
            raise TokenError("invalid_scope", "Scope exceeds the original grant.")
        return self.tokens(
            client.client_id,
            refresh_token.subject,
            scopes or refresh_token.scopes,
            refresh_token.resource,
        )

    async def load_access_token(self, token):
        return None  # The MCP server verifies access tokens itself.

    async def revoke_token(self, token):
        if isinstance(token, RefreshToken):
            self.refresh_tokens.pop(token.token, None)


provider = Provider()


def protect(response: Response) -> Response:
    response.headers["X-Frame-Options"] = "DENY"
    response.headers["Content-Security-Policy"] = "frame-ancestors 'none'"
    return response


async def consent_page(request: Request) -> Response:
    pending = provider.pending.get(request.query_params.get("request", ""))
    if not pending or time.time() - pending["at"] > PENDING_SECONDS:
        return HTMLResponse("This request has expired.", status_code=400)
    client, params = pending["client"], pending["params"]
    hostname = urlparse(str(params.redirect_uri)).hostname
    warning = ""
    if hostname in {"localhost", "127.0.0.1"}:
        warning = "<p><strong>This app runs on your own computer.</strong> Only continue if you started it.</p>"
    requested = params.scopes or SCOPES
    boxes = "".join(
        f'<label><input type="checkbox" name="scope" value="{html.escape(s)}" checked> '
        f"{html.escape(s)}</label><br>"
        for s in requested
        if s in SCOPES
    )
    csrf = secrets.token_urlsafe(24)
    response = HTMLResponse(f"""<!doctype html><title>Allow access</title>
<h1>Allow {html.escape(client.client_name or "this app")} to use your notes?</h1>
<p>Access will be sent to <code>{html.escape(str(params.redirect_uri))}</code></p>
{warning}
<form method="post">
  <input type="hidden" name="request" value="{html.escape(request.query_params["request"])}">
  <input type="hidden" name="csrf" value="{csrf}">
  <p><label>Demo user <input name="user" required></label></p>
  {boxes}
  <button name="action" value="allow">Allow</button>
  <button name="action" value="deny">Deny</button>
</form>""")
    response.set_cookie("csrf", csrf, httponly=True, samesite="strict")
    return protect(response)


async def consent_submit(request: Request) -> Response:
    form = await request.form()
    if not secrets.compare_digest(str(form.get("csrf", "")), request.cookies.get("csrf", "-")):
        return HTMLResponse("Invalid request.", status_code=403)
    pending = provider.pending.pop(str(form.get("request")), None)
    if not pending or time.time() - pending["at"] > PENDING_SECONDS:
        return HTMLResponse("This request has expired.", status_code=400)
    client, params = pending["client"], pending["params"]

    if form.get("action") != "allow":
        target = construct_redirect_uri(
            str(params.redirect_uri), error="access_denied", state=params.state, iss=AUTH_URL
        )
        return RedirectResponse(target, status_code=302)

    # The password check belongs here. Your identity provider handles it in production.
    scopes = [s for s in form.getlist("scope") if s in SCOPES]
    code = provider.issue_code(client, params, str(form["user"]), scopes)
    target = construct_redirect_uri(
        str(params.redirect_uri), code=code, state=params.state, iss=AUTH_URL
    )
    return RedirectResponse(target, status_code=302)


async def jwks(_: Request) -> Response:
    public = json.loads(jwt.algorithms.RSAAlgorithm.to_jwk(private_key.public_key()))
    return JSONResponse({"keys": [{**public, "kid": key_id, "alg": "RS256", "use": "sig"}]})


# AuthSettings keeps the URL exact. A plain AnyHttpUrl adds a trailing slash, and
# clients compare issuers as exact strings, so the connection would be rejected.
issuer = AuthSettings(issuer_url=AUTH_URL, resource_server_url=None).issuer_url

app = Starlette(routes=[
    *create_auth_routes(
        provider,
        issuer,
        client_registration_options=ClientRegistrationOptions(
            enabled=True, valid_scopes=SCOPES, default_scopes=SCOPES
        ),
        revocation_options=RevocationOptions(enabled=True),
    ),
    Route("/consent", consent_page, methods=["GET"]),
    Route("/consent", consent_submit, methods=["POST"]),
    Route("/jwks.json", jwks),
])

"""Starts both servers and runs the real OAuth flow over HTTP."""
import base64
import hashlib
import json
import os
import re
import secrets
import tempfile
import threading
import time
import unittest
from html import unescape
from urllib.parse import parse_qs, urlencode, urlparse

os.environ.update(
    AUTH_URL="http://localhost:9191",
    MCP_URL="http://localhost:8181",
    DATABASE=os.path.join(tempfile.mkdtemp(), "notes.db"),
)

import httpx2
import jwt
import uvicorn
from mcp import Client
from mcp.client.auth import OAuthClientProvider
from mcp.client.streamable_http import streamable_http_client
from mcp.shared.auth import AuthorizationCodeResult, OAuthClientMetadata

import auth_server
import server
from config import AUTH_URL, RESOURCE


def start(app, port):
    instance = uvicorn.Server(uvicorn.Config(app, port=port, log_level="error"))
    threading.Thread(target=instance.run, daemon=True).start()
    while not instance.started:
        time.sleep(0.02)


class MemoryStorage:
    tokens = client_info = None

    async def get_tokens(self):
        return self.tokens

    async def set_tokens(self, tokens):
        self.tokens = tokens

    async def get_client_info(self):
        return self.client_info

    async def set_client_info(self, client_info):
        self.client_info = client_info


class FakeBrowser:
    """Does what a person's browser does on the consent page."""

    def __init__(self, user, scopes=("notes:read", "notes:write"), action="allow"):
        self.user, self.scopes, self.action = user, scopes, action
        self.redirect = None

    async def open(self, url):
        async with httpx2.AsyncClient(follow_redirects=False) as browser:
            response = await browser.get(url)
            while response.status_code in (302, 307):
                response = await browser.get(response.headers["location"])
            page = response.text
            field = lambda name: unescape(re.search(rf'name="{name}" value="([^"]+)"', page).group(1))
            form = {"request": field("request"), "csrf": field("csrf"),
                    "user": self.user, "action": self.action, "scope": list(self.scopes)}
            response = await browser.post(response.url, data=form)
            self.redirect = response.headers["location"]

    async def callback(self):
        query = parse_qs(urlparse(self.redirect).query)
        return AuthorizationCodeResult(
            code=query["code"][0], state=query.get("state", [None])[0], iss=query.get("iss", [None])[0]
        )


def oauth_client(browser):
    auth = OAuthClientProvider(
        server_url=RESOURCE,
        client_metadata=OAuthClientMetadata(
            client_name="Test client",
            redirect_uris=["http://localhost:3333/callback"],
            grant_types=["authorization_code", "refresh_token"],
            response_types=["code"],
        ),
        storage=MemoryStorage(),
        redirect_handler=browser.open,
        callback_handler=browser.callback,
    )
    return Client(streamable_http_client(RESOURCE, http_client=httpx2.AsyncClient(auth=auth)))


def mint(**overrides):
    now = int(time.time())
    claims = {"iss": AUTH_URL, "sub": "mallory", "aud": RESOURCE, "scope": "notes:read notes:write",
              "client_id": "x", "iat": now, "exp": now + 60, **overrides}
    return jwt.encode(claims, auth_server.private_key, algorithm="RS256",
                      headers={"kid": auth_server.key_id})


async def call_with(token):
    headers = {"Authorization": f"Bearer {token}", "Content-Type": "application/json",
               "Accept": "application/json, text/event-stream"}
    body = {"jsonrpc": "2.0", "id": 1, "method": "tools/list", "params": {}}
    async with httpx2.AsyncClient() as http:
        return await http.post(RESOURCE, headers=headers, json=body)


def text(result):
    return json.dumps(result.structured_content) if result.structured_content else result.content[0].text


class Flow(unittest.IsolatedAsyncioTestCase):
    @classmethod
    def setUpClass(cls):
        start(auth_server.app, 9191)
        start(server.app, 8181)

    async def test_no_token_points_to_discovery(self):
        async with httpx2.AsyncClient() as http:
            response = await http.post(RESOURCE, json={})
            self.assertEqual(response.status_code, 401)
            challenge = response.headers["www-authenticate"]
            self.assertIn("resource_metadata=", challenge)
            url = re.search(r'resource_metadata="([^"]+)"', challenge).group(1)
            metadata = (await http.get(url)).json()
            self.assertEqual(metadata["authorization_servers"], [AUTH_URL])
            self.assertEqual(metadata["resource"], RESOURCE)

    async def test_full_oauth_flow_and_tools(self):
        async with oauth_client(FakeBrowser("alice")) as client:
            names = sorted(tool.name for tool in (await client.list_tools()).tools)
            self.assertEqual(names, ["add_note", "delete_note", "list_notes"])
            await client.call_tool("add_note", {"text": "alice's secret"})
            self.assertIn("alice's secret", text(await client.call_tool("list_notes", {})))

    async def test_users_cannot_see_or_delete_each_others_notes(self):
        async with oauth_client(FakeBrowser("carol")) as carol:
            note_id = text(await carol.call_tool("add_note", {"text": "carol's note"}))
            note_id = re.sub(r'.*"result":\s*"|".*', "", note_id) if "{" in note_id else note_id
            async with oauth_client(FakeBrowser("dave")) as dave:
                self.assertNotIn("carol's note", text(await dave.call_tool("list_notes", {})))
                self.assertTrue((await dave.call_tool("delete_note", {"note_id": note_id})).is_error)
            self.assertIn("carol's note", text(await carol.call_tool("list_notes", {})))

    async def test_read_only_grant_cannot_write(self):
        async with oauth_client(FakeBrowser("erin", scopes=["notes:read"])) as client:
            result = await client.call_tool("add_note", {"text": "nope"})
            self.assertTrue(result.is_error)
            self.assertIn("notes:write", result.content[0].text)
            self.assertFalse((await client.call_tool("list_notes", {})).is_error)

    async def test_bad_tokens_are_rejected(self):
        self.assertEqual((await call_with(mint())).status_code, 200)
        cases = {
            "wrong audience": mint(aud="https://other.example/mcp"),
            "wrong issuer": mint(iss="https://evil.example"),
            "expired": mint(exp=int(time.time()) - 5),
            "tampered": mint()[:-4] + "AAAA",
            "unsigned": jwt.encode({"sub": "x"}, None, algorithm="none"),
            "garbage": "not-a-token",
        }
        for name, token in cases.items():
            with self.subTest(name):
                self.assertEqual((await call_with(token)).status_code, 401)

    async def test_token_signed_with_another_key_is_rejected(self):
        from cryptography.hazmat.primitives.asymmetric import rsa
        other = rsa.generate_private_key(public_exponent=65537, key_size=2048)
        now = int(time.time())
        forged = jwt.encode({"iss": AUTH_URL, "sub": "x", "aud": RESOURCE, "exp": now + 60},
                            other, algorithm="RS256", headers={"kid": auth_server.key_id})
        self.assertEqual((await call_with(forged)).status_code, 401)

    async def test_unknown_redirect_uri_cannot_register(self):
        async with httpx2.AsyncClient() as http:
            response = await http.post(f"{AUTH_URL}/register", json={
                "client_name": "Evil", "redirect_uris": ["https://evil.example/callback"],
                "token_endpoint_auth_method": "none", "grant_types": ["authorization_code"],
                "response_types": ["code"],
            })
            self.assertEqual(response.status_code, 400)

    async def test_claude_redirects_can_register_as_public_clients(self):
        uris = ["https://claude.ai/api/mcp/auth_callback", "http://localhost:3118/callback",
                "http://127.0.0.1:51000/callback"]
        async with httpx2.AsyncClient() as http:
            response = await http.post(f"{AUTH_URL}/register", json={
                "client_name": "Claude", "redirect_uris": uris,
                "token_endpoint_auth_method": "none",
                "grant_types": ["authorization_code", "refresh_token"],
                "response_types": ["code"],
            })
            self.assertEqual(response.status_code, 201)
            self.assertNotIn("client_secret", {k for k, v in response.json().items() if v})
            metadata = (await http.get(f"{AUTH_URL}/.well-known/oauth-authorization-server")).json()
            self.assertEqual(metadata["code_challenge_methods_supported"], ["S256"])
            self.assertEqual(metadata["issuer"], AUTH_URL)

    async def test_public_client_completes_pkce_without_a_secret(self):
        redirect = "http://localhost:3118/callback"
        verifier = secrets.token_urlsafe(48)
        challenge = base64.urlsafe_b64encode(
            hashlib.sha256(verifier.encode()).digest()
        ).rstrip(b"=").decode()
        async with httpx2.AsyncClient() as http:
            client = (await http.post(f"{AUTH_URL}/register", json={
                "client_name": "Public", "redirect_uris": [redirect],
                "token_endpoint_auth_method": "none",
                "grant_types": ["authorization_code", "refresh_token"],
                "response_types": ["code"],
            })).json()
            browser = FakeBrowser("pat")
            await browser.open(f"{AUTH_URL}/authorize?" + urlencode({
                "response_type": "code", "client_id": client["client_id"],
                "redirect_uri": redirect, "code_challenge": challenge,
                "code_challenge_method": "S256", "state": "s", "resource": RESOURCE,
            }))
            code = (await browser.callback()).code
            form = {"grant_type": "authorization_code", "code": code, "redirect_uri": redirect,
                    "client_id": client["client_id"], "resource": RESOURCE}
            wrong = await http.post(f"{AUTH_URL}/token", data={**form, "code_verifier": "x" * 48})
            self.assertEqual(wrong.status_code, 400)
            right = await http.post(f"{AUTH_URL}/token", data={**form, "code_verifier": verifier})
            self.assertEqual(right.status_code, 200, right.text)
            self.assertEqual((await call_with(right.json()["access_token"])).status_code, 200)

    async def test_older_protocol_handshake_still_works(self):
        headers = {"Authorization": f"Bearer {mint()}", "Content-Type": "application/json",
                   "Accept": "application/json, text/event-stream"}
        body = {"jsonrpc": "2.0", "id": 1, "method": "initialize", "params": {
            "protocolVersion": "2025-11-25", "capabilities": {},
            "clientInfo": {"name": "older", "version": "1"}}}
        async with httpx2.AsyncClient() as http:
            response = await http.post(RESOURCE, headers=headers, json=body)
            self.assertEqual(response.status_code, 200, response.text)
            self.assertEqual(response.json()["result"]["protocolVersion"], "2025-11-25")

    async def test_consent_needs_the_csrf_cookie(self):
        async with httpx2.AsyncClient() as http:
            response = await http.post(f"{AUTH_URL}/consent", data={"request": "x", "csrf": "forged"})
            self.assertEqual(response.status_code, 403)

    async def test_wrong_host_header_is_rejected(self):
        headers = {"Host": "evil.example", "Authorization": f"Bearer {mint()}",
                   "Content-Type": "application/json", "Accept": "application/json, text/event-stream"}
        async with httpx2.AsyncClient() as http:
            response = await http.post(RESOURCE, headers=headers, json={})
            self.assertIn(response.status_code, (403, 421))

    async def test_authorization_code_works_once(self):
        provider = auth_server.provider
        client = type("C", (), {"client_id": "c"})()
        params = type("P", (), {"code_challenge": "x", "redirect_uri": "http://localhost:1/cb",
                                "redirect_uri_provided_explicitly": True, "resource": RESOURCE})()
        code = provider.issue_code(client, params, "u", ["notes:read"])
        loaded = await provider.load_authorization_code(client, code)
        await provider.exchange_authorization_code(client, loaded)
        self.assertIsNone(await provider.load_authorization_code(client, code))

    async def test_refresh_tokens_rotate(self):
        provider = auth_server.provider
        client = type("C", (), {"client_id": "c"})()
        first = provider.tokens("c", "u", ["notes:read"], RESOURCE)
        loaded = await provider.load_refresh_token(client, first.refresh_token)
        second = await provider.exchange_refresh_token(client, loaded, ["notes:read"])
        self.assertNotEqual(first.refresh_token, second.refresh_token)
        self.assertIsNone(await provider.load_refresh_token(client, first.refresh_token))


if __name__ == "__main__":
    unittest.main()