Building a Token Server

This page is for whoever is implementing the token server — usually the same team that owns the upstream API integration. If you’re just sending requests through Proxy Hopper and someone else runs the token server for your target, see Authenticating instead — that page covers a different thing, authenticating your requests to Proxy Hopper, not the token server that Proxy Hopper itself calls.

What you’re building

A token server is a small HTTP service with two endpoints:

POST /token — given a target name, a proxy IP, and an opaque cursor, return the headers to inject into the upstream request, when they expire, and an updated cursor.
GET /health — return 2xx once ready. Proxy Hopper calls this at startup before pre-warming tokens for every configured (target, IP) pair.

You can implement this contract by hand in any language or framework — see the wire contract below. In Python, the proxy-hopper-token-server library gets you a tested implementation of both endpoints for free: you write one async method, TokenProvider.get_token(), and it handles the FastAPI app, request validation, timeout enforcement, and error responses.

Scope credentials per IP

Read this before writing get_token() — it’s the one rule that matters more than everything else on this page. Proxy Hopper calls your token server separately for every (target, proxy-IP) pair — the request body always includes req.ip. This exists so that each proxy IP can carry its own distinct credential, not so every IP conveniently shares one. The reason: Proxy Hopper’s entire purpose is to spread requests across a rotating pool of IPs so a target can’t easily tell they all come from the same caller. If get_token() ignores req.ip and returns the same static token for every IP — a single OAuth client-credentials token fetched once and handed back for every request, for example — Proxy Hopper will still cache it separately per IP, but the credential value itself becomes the one constant thread linking every IP in the pool back to the same caller. That silently defeats the whole point of the pool, and it’s a much easier signal for a target to key on than IP address alone.

Concretely: if the upstream API supports per-session or per-login credentials (most OAuth and session-cookie flows do), get_token() should acquire a separate login/session per req.ip, not fetch one token and reuse it for every request that happens to hit this endpoint. Use req.cursor to hold per-IP session state (it’s already scoped per (target, ip) — see the cursor mechanism), and see IP-pinned acquisition below for routing the login call itself through the same IP.

This doesn’t apply to upstream auth that’s genuinely IP-agnostic by nature — a single organisation-wide API key with no per-session concept, for instance. There, one shared credential is correct because there’s nothing to scope: the upstream can’t distinguish sessions either way. The mistake is only returning a shared credential for an auth mechanism that does support scoping, and not taking advantage of it.

Quick start (using the library)

pip install proxy-hopper-token-server

1. Implement a TokenProvider:

# mytokens.py
from datetime import UTC, datetime, timedelta
from proxy_hopper_token_server import TokenProvider, TokenRequest, TokenResponse

class MyTokens(TokenProvider):
    async def get_token(self, req: TokenRequest) -> TokenResponse:
        # req.target   — the Proxy Hopper target name this token is for
        # req.ip, req.port — the upstream proxy IP that will use this token
        # req.cursor   — opaque dict you returned last call for this (target, ip); {} on first call
        # req.profile  — the browser fingerprint Proxy Hopper is using for this IP
        token = await fetch_a_real_token()
        return TokenResponse(
            headers={"Authorization": f"Bearer {token}"},
            expires_at=datetime.now(UTC) + timedelta(hours=1),
            cursor=req.cursor,   # unchanged — return updated state here if you need it
        )

provider = MyTokens()

2. Run it:

ph-token-server start mytokens:provider --port 9000

3. Point Proxy Hopper at it in config.yaml:

server:
  authServer:
    url: "http://localhost:9000"

targets:
  - name: my-api
    regex: 'api\.example\.com'
    ipPool: my-pool
    authManaged: true

That’s the whole integration — Proxy Hopper calls POST /token before forwarding through this target, caches the response per (target, ip), and refreshes it before expires_at.

The wire contract

If you’re implementing this from scratch (a different language, or embedding it in an existing service), your server needs to expose exactly this: POST /token — request body:

{
  "target": "my-api",
  "ip": "203.0.113.10",
  "port": 3128,
  "cursor": {},
  "profile": {
    "user_agent": "Mozilla/5.0 ...",
    "accept": "text/html",
    "accept_language": "en-US",
    "accept_encoding": "gzip",
    "extra": {}
  },
  "proxy_url": "http://proxy-hopper:8080"
}

proxy_url is only present when server.authServer.exposeProxyUrl: true is set (the default) — see IP-pinned acquisition below for why you’d want it. Response body:

{
  "headers": {"Authorization": "Bearer eyJ..."},
  "expires_at": "2026-08-02T12:00:00+00:00",
  "cursor": {}
}

Field	Description
`headers`	Key-value pairs injected into the upstream request, replacing any same-named header the client sent
`expires_at`	ISO-8601 UTC timestamp. Proxy Hopper proactively refreshes before this, per `refreshThresholdSeconds`
`cursor`	Echoed back on the next `/token` call for this `(target, ip)` pair — see the cursor mechanism

GET /health — must return 2xx once ready. Using the library, TokenServer/create_app build exactly this contract for you — you only ever touch TokenRequest/TokenResponse dataclasses, never raw HTTP.

The cursor mechanism

cursor is an opaque JSON object Proxy Hopper stores per (target, ip) pair and echoes back on every subsequent call — a small key-value slot for anything you need to correlate calls (a refresh token, a session ID, a call counter) without running your own database. Return the incoming cursor unchanged if you don’t need it.

async def get_token(self, req: TokenRequest) -> TokenResponse:
    refresh_token = req.cursor.get("refresh_token")
    if refresh_token:
        access, new_refresh, expires = await exchange_refresh_token(refresh_token)
    else:
        access, new_refresh, expires = await fresh_login()
    return TokenResponse(
        headers={"Authorization": f"Bearer {access}"},
        expires_at=expires,
        cursor={"refresh_token": new_refresh},
    )

IP-pinned acquisition

Some auth endpoints tie a session or token to the IP address that requested it. If your get_token() needs to make its own outbound call (e.g. logging in) from the same proxy IP that will later use the resulting token, route that call back through Proxy Hopper on a pinned IP using ProxyHopperClient:

from proxy_hopper_token_server import ProxyHopperClient, TokenProvider, TokenRequest, TokenResponse

class MyTokens(TokenProvider):
    async def get_token(self, req: TokenRequest) -> TokenResponse:
        client = ProxyHopperClient(proxy_url=req.proxy_url)   # requires exposeProxyUrl: true
        resp = await client.post(
            "https://auth.example.com/login",
            via_ip=f"{req.ip}:{req.port}",
            headers={"User-Agent": req.profile.user_agent},
            data=b'{"user": "...", "pass": "..."}',
        )
        body = await resp.read()
        ...

ProxyHopperClient sends the request to Proxy Hopper’s own address using the same header-based forwarding every other client uses (X-Proxy-Hopper-Target + X-ProxyHopper-Force-IP) — it does not use a classic HTTP-proxy or CONNECT-tunnel request, since Proxy Hopper’s core doesn’t implement those modes. req.proxy_url is only populated when exposeProxyUrl: true; without it, make the call with a plain HTTP client instead (no IP pinning). client.post() / .get() return a ProxyHopperResponse (.status, .headers, await .read(), .text()) with the body fully buffered — safe to read after the call returns, and safe to read more than once.

See auckland_council.py in the repo for a complete, real-world TokenProvider that uses IP-pinned acquisition and carries a session cookie in the cursor across refreshes.

Common patterns

OAuth2 client credentials

Client-credentials tokens authenticate your application, not a session — there’s no login and no per-IP concept in this grant type, so every IP ends up presenting the same client identity to the upstream regardless of what get_token() does with req.ip. That’s the “genuinely IP-agnostic” case from Scope credentials per IP — nothing to fix here. If the target instead supports session or refresh-token based auth, prefer one of the two patterns below, which do give each IP a distinct credential.

import httpx, os

class OAuthTokens(TokenProvider):
    async def get_token(self, req: TokenRequest) -> TokenResponse:
        async with httpx.AsyncClient() as client:
            resp = await client.post(
                "https://auth.example.com/oauth/token",
                data={
                    "grant_type": "client_credentials",
                    "client_id": os.environ["OAUTH_CLIENT_ID"],
                    "client_secret": os.environ["OAUTH_CLIENT_SECRET"],
                    "scope": f"proxy:{req.target}",
                },
            )
            resp.raise_for_status()
            data = resp.json()
        return TokenResponse(
            headers={"Authorization": f"Bearer {data['access_token']}"},
            expires_at=datetime.now(UTC) + timedelta(seconds=data["expires_in"]),
            cursor=req.cursor,
        )

Refresh-token exchange (state carried in the cursor)

This pattern gives each IP its own credential without ever branching on req.ip directly — it relies on req.cursor already being scoped per (target, ip) by Proxy Hopper. The first call for a given IP has an empty cursor and mints a fresh token pair for that IP alone; every later call for that same IP refreshes from its own stored refresh token. No two IPs ever share a refresh token or the access token derived from it.

class RefreshTokens(TokenProvider):
    async def get_token(self, req: TokenRequest) -> TokenResponse:
        refresh_token = req.cursor.get("refresh_token")
        async with httpx.AsyncClient() as client:
            if refresh_token:
                resp = await client.post(TOKEN_URL, data={
                    "grant_type": "refresh_token", "refresh_token": refresh_token,
                })
            else:
                resp = await client.post(TOKEN_URL, data={
                    "grant_type": "client_credentials",
                    "client_id": os.environ["OAUTH_CLIENT_ID"],
                    "client_secret": os.environ["OAUTH_CLIENT_SECRET"],
                })
            data = resp.json()
        return TokenResponse(
            headers={"Authorization": f"Bearer {data['access_token']}"},
            expires_at=datetime.now(UTC) + timedelta(seconds=data["expires_in"]),
            cursor={"refresh_token": data["refresh_token"]},
        )

If the upstream needs a session cookie instead of a bearer token, return it as a Cookie header instead of Authorization. Like the refresh-token pattern, this is correctly scoped per IP for free — Proxy Hopper calls get_token() independently for each IP, so each IP performs its own /login and gets back its own session cookie, never a shared one:

class CookieTokens(TokenProvider):
    async def get_token(self, req: TokenRequest) -> TokenResponse:
        async with httpx.AsyncClient() as client:
            resp = await client.post(
                "https://api.example.com/login",
                json={"username": os.environ["API_USER"], "password": os.environ["API_PASS"]},
            )
            session_cookie = resp.cookies["session"]
        return TokenResponse(
            headers={"Cookie": f"session={session_cookie}"},
            expires_at=datetime.now(UTC) + timedelta(hours=1),
            cursor=req.cursor,
        )

Multiple providers, one server

Route by target name by passing a dict[str, TokenProvider] instead of a single provider:

TokenServer({
    "target-a": ProviderA(),
    "target-b": ProviderB(),
})

A request for a target with no matching key returns 404.

Testing your provider

TokenServer.build_app() returns a plain FastAPI app — test it with fastapi.testclient.TestClient, no network and no running Proxy Hopper instance needed:

from fastapi.testclient import TestClient
from proxy_hopper_token_server import TokenServer
from mytokens import provider

def test_token_issued():
    app = TokenServer(provider).build_app()
    with TestClient(app) as client:
        resp = client.post("/token", json={
            "target": "my-api", "ip": "1.2.3.4", "port": 3128, "cursor": {},
            "profile": {"user_agent": "", "accept": "", "accept_language": "", "accept_encoding": ""},
        })
    assert resp.status_code == 200
    assert "Authorization" in resp.json()["headers"]

Failure behavior

Any exception raised from get_token() becomes a 500 with {"error": "provider_error", "detail": str(exc)}. Exceeding the configured --timeout becomes a 504 with {"error": "timeout", ...}. Proxy Hopper treats repeated failures as AUTH_BROKEN for that IP — see Token Server (Admin) — Broken-state and recovery.

CLI reference

ph-token-server start IMPORT_PATH [OPTIONS]

  IMPORT_PATH                'module.path:AttributeName' — resolves to a
                              TokenServer instance, TokenProvider instance,
                              or TokenProvider subclass (instantiated with no args)

  --host TEXT                 Interface to bind [default: 0.0.0.0]
  --port INT                  Port to listen on [default: 9000]
  --workers INT                uvicorn worker processes [default: 1]
  --log-level CHOICE          trace|debug|info|warning|error [default: info]
  --timeout FLOAT              Hard timeout (seconds) per get_token() call [default: 30.0]

CLI flags override whatever the resolved instance was constructed with.

ph-token-server start myapp.tokens:MyProvider
ph-token-server start myapp.tokens:provider --port 9001
ph-token-server start myapp.tokens:server --workers 4

Hosting behind your own ASGI server

For local development with autoreload, or to embed the token endpoints in a larger FastAPI app, use TokenServer.build_app() to get the plain ASGI app instead of calling .run():

# main.py
from proxy_hopper_token_server import TokenServer
from mytokens import provider

app = TokenServer(provider).build_app()

uvicorn main:app --reload --port 9000

Full runnable example

The repo includes an end-to-end example with a from-scratch FastAPI token server, Docker Compose wiring, and recipes for swapping in real OAuth2/session-cookie/rotating-key auth:

examples/token-server/

Complete Docker Compose stack — token server + Proxy Hopper — with a signed-JWT demo you swap for your real auth mechanism.

auckland_council.py

A real-world provider using IP-pinned acquisition and a cursor-carried session cookie, built on this library.

Getting Started

Concepts

For Administrators

For Developers

For Contributors

Building a Token Server

What you’re building

Scope credentials per IP

Quick start (using the library)

The wire contract

The cursor mechanism

IP-pinned acquisition

Common patterns

OAuth2 client credentials

Refresh-token exchange (state carried in the cursor)

Multiple providers, one server

Testing your provider

Failure behavior

CLI reference

Hosting behind your own ASGI server

Full runnable example

examples/token-server/

auckland_council.py

Next steps

Managed Auth

Token Server (Admin)

​What you’re building

​Scope credentials per IP

​Quick start (using the library)

​The wire contract

​The cursor mechanism

​IP-pinned acquisition

​Common patterns

​OAuth2 client credentials

​Refresh-token exchange (state carried in the cursor)

​Session-cookie auth

​Multiple providers, one server

​Testing your provider

​Failure behavior

​CLI reference

​Hosting behind your own ASGI server

​Full runnable example

examples/token-server/

auckland_council.py

​Next steps

Managed Auth

Token Server (Admin)

What you’re building

Scope credentials per IP

Quick start (using the library)

The wire contract

The cursor mechanism

IP-pinned acquisition

Common patterns

OAuth2 client credentials

Refresh-token exchange (state carried in the cursor)

Session-cookie auth

Multiple providers, one server

Testing your provider

Failure behavior

CLI reference

Hosting behind your own ASGI server

Full runnable example

Next steps