> ## Documentation Index
> Fetch the complete documentation index at: https://proxy-hopper.io/llms.txt
> Use this file to discover all available pages before exploring further.

# Building a Token Server

> Implement the token server that backs Proxy Hopper's managed auth feature — fetch and refresh upstream credentials so your application never has to.

<Note>
  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](/developers/authenticating) instead — that page covers a different thing, authenticating your requests *to* Proxy Hopper, not the token server that Proxy Hopper itself calls.
</Note>

## 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](#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.

<Warning>
  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](#the-cursor-mechanism)), and see [IP-pinned acquisition](#ip-pinned-acquisition) below for routing the login call itself through the same IP.
</Warning>

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)

```bash theme={}
pip install proxy-hopper-token-server
```

**1. Implement a `TokenProvider`:**

```python theme={}
# 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:**

```bash theme={}
ph-token-server start mytokens:provider --port 9000
```

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

```yaml theme={}
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:

```json theme={}
{
  "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](#ip-pinned-acquisition) below for why you'd want it.

Response body:

```json theme={}
{
  "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](#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.

```python theme={}
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`:

```python theme={}
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.

<Note>
  See [`auckland_council.py`](https://github.com/cams-data/proxy-hopper-v2/blob/next/examples/token-server/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.
</Note>

## Common patterns

### OAuth2 client credentials

<Note>
  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](#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.
</Note>

```python theme={}
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.

```python theme={}
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"]},
        )
```

### Session-cookie auth

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:

```python theme={}
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:

```python theme={}
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:

```python theme={}
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](/admin/token-server/overview#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.

```bash theme={}
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()`:

```python theme={}
# main.py
from proxy_hopper_token_server import TokenServer
from mytokens import provider

app = TokenServer(provider).build_app()
```

```bash theme={}
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:

<CardGroup cols={2}>
  <Card title="examples/token-server/" icon="key" href="https://github.com/cams-data/proxy-hopper-v2/tree/next/examples/token-server">
    Complete Docker Compose stack — token server + Proxy Hopper — with a signed-JWT demo you swap for your real auth mechanism.
  </Card>

  <Card title="auckland_council.py" icon="cookie" href="https://github.com/cams-data/proxy-hopper-v2/blob/next/examples/token-server/auckland_council.py">
    A real-world provider using IP-pinned acquisition and a cursor-carried session cookie, built on this library.
  </Card>
</CardGroup>

## Next steps

<CardGroup cols={2}>
  <Card title="Managed Auth" icon="key" href="/concepts/managed-auth">
    The concept guide — caching, refresh timing, and broken-state recovery.
  </Card>

  <Card title="Token Server (Admin)" icon="server" href="/admin/token-server/overview">
    Deployment, configuration fields, and operational metrics/logs.
  </Card>
</CardGroup>