Notebooks · In-depth guides · Guide 05

Connections and OAuth

Name: SHIPIT Agent
Author: SHIPIT

A connection is a service your agent acts in on someone's behalf: Linear, GitHub, Slack, Notion. In shipit every connector is an MCP server from a catalog. Some read a key from the environment; the hosted ones use…

Open in Colab View on GitHub 17 code cells · 15 with saved output

Run end to end on shipit-agent 3.0 with real model calls (Haiku 5.5 on Anthropic, Gemma 4 31B on Bedrock). Every output below is from that run.

1 · Models

In [1]

python

import os

from shipit_agent import Agent
from shipit_agent.llms import build_llm_from_env
from shipit_agent.llms.factory import build_llm_from_settings

os.environ["SHIPIT_ANTHROPIC_MODEL"] = "claude-haiku-5-5"
haiku = build_llm_from_env("anthropic")
gemma = build_llm_from_settings({"model": "google.gemma-4-31b"}, provider="bedrock")

2 · The catalog

In [2]

python

from collections import Counter

from shipit_agent import list_connectors

connectors = list_connectors()
print(len(connectors), "connectors")
print("by category:", dict(Counter(c.category for c in connectors)))
print("by auth:    ", dict(Counter("oauth" if c.uses_oauth else c.auth for c in connectors)))

Out

28 connectors
by category: {'data': 6, 'productivity': 5, 'automation': 5, 'developer': 6, 'communication': 3, 'business': 3}
by auth:     {'env': 8, 'oauth': 13, 'none': 7}

In [3]

python

for c in connectors:
    how = "hosted" if c.url else "stdio "
    auth = "per-user OAuth" if c.uses_oauth else (", ".join(c.required_env) or "no key")
    print(f"{c.name:14} {how}  {auth:28} {c.description}")

Out

airtable stdio AIRTABLE_API_KEY Airtable bases, tables and records.
asana hosted per-user OAuth Asana tasks, projects and portfolios.
brave-search stdio BRAVE_API_KEY Web search via the Brave Search API.
clickup hosted per-user OAuth ClickUp tasks, lists, docs and spaces.
cloudflare hosted per-user OAuth Cloudflare Workers, KV, R2, D1 and observability.
confluence hosted per-user OAuth Confluence pages and spaces (same grant as Jira).
discord stdio DISCORD_TOKEN Post, read and manage Discord channels and threads.
fetch stdio no key Fetch a URL and return page content as markdown.
filesystem stdio no key Read/write files under the directories you pass as args.
gdrive stdio GDRIVE_CREDENTIALS_PATH Search and read Google Drive files.
github hosted per-user OAuth Repos, issues, PRs, code search and reviews on GitHub.
gitlab stdio GITLAB_PERSONAL_ACCESS_TOKEN GitLab projects, issues and merge requests.
google-maps stdio GOOGLE_MAPS_API_KEY Places, directions and geocoding via Google Maps.
intercom hosted per-user OAuth Intercom conversations, contacts and help articles.
jira hosted per-user OAuth Jira issues, JQL search, transitions and sprints.
linear hosted per-user OAuth Linear issues, cycles, projects and comments.
memory stdio no key Persistent knowledge-graph memory across runs.
notion hosted per-user OAuth Search, read and write Notion pages and databases.
paypal hosted per-user OAuth PayPal orders, invoices and transactions.
playwright stdio no key Drive a browser: navigate, click, type, screenshot.
postgres stdio no key Query PostgreSQL (read-only). Pass the connection URL as an arg.
puppeteer stdio no key Headless browser: navigate, screenshot, interact.
sentry hosted per-user OAuth Sentry issues, events and stack traces.
slack stdio SLACK_BOT_TOKEN, SLACK_TEAM_ID Post and read Slack messages, channels and search.
sqlite stdio no key Query a SQLite database file passed as an arg.
square hosted per-user OAuth Square payments, catalog and customers.
stripe hosted STRIPE_SECRET_KEY Stripe customers, charges, invoices and subscriptions.
todoist hosted per-user OAuth Todoist tasks, projects and labels.

3 · One connector's manifest

Each connector is one declarative manifest: where it lives, how it authenticates, and the scopes it asks for.

In [4]

python

linear = next(c for c in connectors if c.name == "linear")
for field in ("name", "category", "kind", "url", "auth", "uses_oauth", "oauth", "scopes", "suggest"):
    print(f"{field:11} {getattr(linear, field)!r}")

Out

name        'linear'
category    'developer'
kind        'hosted'
url         'https://mcp.linear.app/sse'
auth        'oauth'
uses_oauth  True
oauth       'linear'
scopes      ['read', 'write', 'issues:create']
suggest     ['linear', 'issue', 'ticket']

4 · Missing keys fail early, with a clear message

connect() checks a connector's requirements before anything starts.

In [5]

python

from shipit_agent import connect

for name in ("linear", "airtable"):
    try:
        connect(name)
    except Exception as error:
        print(f"{name}: {type(error).__name__}: {error}\n")

Out

linear: ValueError: Connector 'linear' is per-user OAuth — pass token=<access token>. Run its OAuth flow (OAUTH_PRESETS['linear']) and store the token per user; refresh it when token_is_expired().

airtable: ValueError: Connector 'airtable' needs env var(s): AIRTABLE_API_KEY. Pass them via env={...} or export them.

5 · A local OAuth provider

A real provider (Linear, GitHub) has an authorize page where the user signs in and approves, and a token endpoint. This stand-in approves every request at once, issues one-time codes, swaps them for tokens, and can refresh or revoke them. It runs in a background thread on a free port.

In [6]

python

import json
import secrets
import threading
from http.server import BaseHTTPRequestHandler, HTTPServer
from urllib.parse import parse_qs, urlencode, urlparse

class Provider:
    codes: dict[str, str] = {}
    refresh_tokens: set[str] = set()
    calls: Counter = Counter()

class Handler(BaseHTTPRequestHandler):
    def log_message(self, *args):  # keep the notebook quiet
        pass

    def do_GET(self):  # the authorize page: the user approves, we redirect back with a code
        q = parse_qs(urlparse(self.path).query)
        code = secrets.token_urlsafe(8)
        Provider.codes[code] = q["client_id"][0]
        Provider.calls["authorize"] += 1
        self.send_response(302)
        self.send_header("Location", f"{q['redirect_uri'][0]}?{urlencode({'code': code, 'state': q['state'][0]})}")
        self.end_headers()

    def do_POST(self):  # the token endpoint
        form = {k: v[0] for k, v in parse_qs(self.rfile.read(int(self.headers["content-length"])).decode()).items()}
        Provider.calls[form["grant_type"]] += 1
        ok = (form["grant_type"] == "authorization_code" and Provider.codes.pop(form.get("code", ""), None)) or \
             (form["grant_type"] == "refresh_token" and form.get("refresh_token") in Provider.refresh_tokens)
        if not ok:
            self.send_response(400)
            self.end_headers()
            self.wfile.write(b'{"error": "invalid_grant"}')
            return
        refresh = secrets.token_urlsafe(12)
        Provider.refresh_tokens.add(refresh)
        body = {"access_token": "at_" + secrets.token_urlsafe(12), "refresh_token": refresh,
                "token_type": "Bearer", "expires_in": 3600}
        self.send_response(200)
        self.send_header("content-type", "application/json")
        self.end_headers()
        self.wfile.write(json.dumps(body).encode())

server = HTTPServer(("127.0.0.1", 0), Handler)
threading.Thread(target=server.serve_forever, daemon=True).start()
BASE = f"http://127.0.0.1:{server.server_port}"
print("provider running at", BASE)

Out

provider running at http://127.0.0.1:61173

6 · The flow by hand, step 1: send the user to sign in

OAuthHelper is the authorization-code flow on its own. create_authorization_url makes a one-time state nonce, saves it with your payload (here, which user is connecting), and returns the URL to redirect to.

In [7]

python

from pathlib import Path

from shipit_agent import FileOAuthStateStore, OAuthClientConfig, OAuthHelper

config = OAuthClientConfig(
    client_id="demo-client", client_secret="demo-secret",
    redirect_uri="https://app.example.com/oauth/callback",
    scopes=["read", "write"],
    authorize_url=f"{BASE}/authorize", token_url=f"{BASE}/token",
)
oauth = OAuthHelper(config, state_store=FileOAuthStateStore("oauth_state.json"))

start = oauth.create_authorization_url(state_payload={"user_id": "u_812"})
print(start["url"].replace(BASE, "<provider>"))

Out

<provider>/authorize?client_id=demo-client&redirect_uri=https%3A%2F%2Fapp.example.com%2Foauth%2Fcallback&scope=read+write&response_type=code&state=AZ2tTmoqPLj76QEDb7hS9aNxciSDK80Q

7 · Step 2: the callback brings a code

In a web app the browser comes back to your redirect URI. Here we follow the provider's redirect by hand and read the code and state from it.

In [8]

python

import http.client

def sign_in(url: str) -> dict[str, str]:
    """What the browser does: open the authorize page and land on the callback."""
    parsed = urlparse(url)
    conn = http.client.HTTPConnection(parsed.hostname, parsed.port)
    conn.request("GET", f"{parsed.path}?{parsed.query}")
    location = conn.getresponse().getheader("Location")
    return {k: v[0] for k, v in parse_qs(urlparse(location).query).items()}

callback = sign_in(start["url"])
print({k: v[:6] + "…" for k, v in callback.items()})

Out

{'code': 'acH2La…', 'state': 'AZ2tTm…'}

8 · Step 3: exchange the code for tokens

Passing state checks it against the saved nonce: the defence against login CSRF. The result is stamped with expires_at.

In [9]

python

tokens = oauth.exchange_code(code=callback["code"], state=callback["state"])
print({k: (v[:9] + "…" if isinstance(v, str) and len(v) > 12 else v) for k, v in tokens.items()})

Out

{'access_token': 'at_EczvJX…', 'refresh_token': 'E0uvWaGGn…', 'token_type': 'Bearer', 'expires_in': 3600, 'expires_at': 1794145995}

9 · A state works once

The nonce is consumed on success, so a replayed callback is refused before any request is made.

In [10]

python

try:
    oauth.exchange_code(code=callback["code"], state=callback["state"])
except ValueError as error:
    print("refused:", error)

Out

refused: Invalid or expired OAuth state — possible CSRF or replay.

10 · Refresh

Access tokens are short-lived; the refresh token gets a new one. token_is_expired says when.

In [11]

python

from shipit_agent.integrations.oauth import token_is_expired

print("expired now?", token_is_expired(tokens))
renewed = oauth.refresh_token(tokens["refresh_token"])
print("new access token differs:", renewed["access_token"] != tokens["access_token"])
print("provider calls so far:", dict(Provider.calls))

Out

expired now? False
new access token differs: True
provider calls so far: {'authorize': 1, 'authorization_code': 1, 'refresh_token': 1}

11 · `OAuthCredentialManager`: the whole flow, per user

In an app you don't juggle tokens yourself. The manager keeps one token per (user, connector) in a store, and refreshes it when it's near expiry. Register the connector's OAuth helper once.

In [12]

python

from shipit_agent.connectors import FileTokenStore, OAuthCredentialManager

manager = OAuthCredentialManager(FileTokenStore("tokens"))
manager.register("linear", OAuthHelper(config, state_store=FileOAuthStateStore("oauth_state.json")))

url = manager.authorize_url(user="ada", connector="linear")       # your "Connect Linear" button
back = sign_in(url)                                                  # Ada approves
manager.complete(connector="linear", code=back["code"], state=back["state"])   # your callback route

print("status:", manager.status(user="ada", connector="linear"))
print("ada's connections:", manager.connections(user="ada"))
print("bob's connections:", manager.connections(user="bob"))

Out

status: connected
ada's connections: {'linear': 'connected'}
bob's connections: {}

12 · Tokens that refresh themselves

access_token() returns a valid token, refreshing it first if it's expired. Here Ada's stored token is aged past its expiry, and the next call refreshes it.

In [13]

python

import time

first = manager.access_token(user="ada", connector="linear")
stored = manager.store.get("ada", "linear")
manager.store.set("ada", "linear", {**stored, "expires_at": time.time() - 10})

before = Provider.calls["refresh_token"]
second = manager.access_token(user="ada", connector="linear")
print("refreshed:", second != first, "| refresh calls:", Provider.calls["refresh_token"] - before)
print("on disk:", sorted(p.name for p in Path("tokens").rglob("*") if p.is_file()))

Out

refreshed: True | refresh calls: 1
on disk: ['linear.json']

13 · A user's token inside a tool

Your own tools can act as the signed-in user: resolve their token at call time, never store it in the prompt. The agent's session_id carries who it's working for.

In [14]

python

from shipit_agent.tools import FunctionTool

def my_issues(cycle: str = "current") -> str:
    """List the signed-in user's Linear issues for a cycle."""
    token = manager.access_token(user="ada", connector="linear")
    # A real tool would call Linear's API with: Authorization: Bearer <token>
    return json.dumps({"cycle": cycle, "auth": token[:6] + "…", "issues": [
        {"id": "ENG-412", "title": "Retry webhooks on 5xx", "state": "In review"},
        {"id": "ENG-418", "title": "Rotate the signing key", "state": "Todo"},
    ]})

agent = Agent(llm=haiku, prompt="You help Ada with her Linear work. Be brief.",
              tools=[FunctionTool.from_callable(my_issues, read_only=True)])
print(agent.run("What's on my plate this cycle, and what's closest to done?").output)

Out

You have two issues this cycle:

- **ENG-412, Retry webhooks on 5xx**: In review. This is closest to done.
- **ENG-418, Rotate the signing key**: Todo. Not started.

14 · When a token is revoked

If the user revokes access at the provider, refresh fails and the manager raises NeedsReconnect: your app shows "Reconnect Linear" instead of making calls that fail silently.

In [15]

python

Provider.refresh_tokens.clear()            # the user revoked access at the provider
stored = manager.store.get("ada", "linear")
manager.store.set("ada", "linear", {**stored, "expires_at": time.time() - 10})
try:
    manager.access_token(user="ada", connector="linear")
except Exception as error:
    print(type(error).__name__, "->", error)

Out

NeedsReconnect -> Refreshing 'linear' for ada failed: HTTP Error 400: Bad Request

15 · The same tool on Gemma 4

After Ada reconnects, Gemma 4 31B (Bedrock) runs the same tool.

In [16]

python

back = sign_in(manager.authorize_url(user="ada", connector="linear"))
manager.complete(connector="linear", code=back["code"], state=back["state"])

agent = Agent(llm=gemma, prompt="You help Ada with her Linear work. Be brief.",
              tools=[FunctionTool.from_callable(my_issues, read_only=True)])
print(agent.run("What's on my plate this cycle?").output)

Out

You have two issues this cycle:
- **ENG-412**: Retry webhooks on 5xx (In review)
- **ENG-418**: Rotate the signing key (Todo)

16 · In a web app

The same three calls map onto two routes and a per-request agent. With FastAPI:

@app.get("/connect/linear")
def connect_linear(user=Depends(current_user)):
    return RedirectResponse(manager.authorize_url(user=user.id, connector="linear"))

@app.get("/oauth/callback")
def callback(code: str, state: str):
    manager.complete(connector="linear", code=code, state=state)
    return RedirectResponse("/settings/connections")

@app.post("/ask")
def ask(body: Question, user=Depends(current_user)):
    agent = Agent(llm=llm, mcps=[manager.connect("linear", user=user.id)])
    return {"answer": agent.run(body.text).output}

manager.connect(connector, user=...) returns the hosted MCP server with that user's token attached.

17 · Clean up

In [17]

python

import shutil

server.shutdown()
shutil.rmtree("tokens", ignore_errors=True)
Path("oauth_state.json").unlink(missing_ok=True)

18 · Summary

list_connectors() is the catalog; each connector says where it lives and how it authenticates.
connect() fails early and clearly when a key or token is missing.
OAuthHelper is the authorization-code flow; state is checked and consumed, so replays are refused.
OAuthCredentialManager keeps per-user tokens, refreshes them, and raises NeedsReconnect when it can't.
Tools resolve the user's token at call time; manager.connect() attaches a hosted connector per user.

Next: Guide 06 — RAG.

Notebooks · In-depth guides · Guide 05

Connections and OAuth

Open in Colab View on GitHub 17 code cells · 15 with saved output

Run end to end on shipit-agent 3.0 with real model calls (Haiku 5.5 on Anthropic, Gemma 4 31B on Bedrock). Every output below is from that run.

1 · Models

In [1]

python

import os

from shipit_agent import Agent
from shipit_agent.llms import build_llm_from_env
from shipit_agent.llms.factory import build_llm_from_settings

os.environ["SHIPIT_ANTHROPIC_MODEL"] = "claude-haiku-5-5"
haiku = build_llm_from_env("anthropic")
gemma = build_llm_from_settings({"model": "google.gemma-4-31b"}, provider="bedrock")

2 · The catalog

In [2]

python

from collections import Counter

from shipit_agent import list_connectors

connectors = list_connectors()
print(len(connectors), "connectors")
print("by category:", dict(Counter(c.category for c in connectors)))
print("by auth:    ", dict(Counter("oauth" if c.uses_oauth else c.auth for c in connectors)))

Out

28 connectors
by category: {'data': 6, 'productivity': 5, 'automation': 5, 'developer': 6, 'communication': 3, 'business': 3}
by auth:     {'env': 8, 'oauth': 13, 'none': 7}

In [3]

python

for c in connectors:
    how = "hosted" if c.url else "stdio "
    auth = "per-user OAuth" if c.uses_oauth else (", ".join(c.required_env) or "no key")
    print(f"{c.name:14} {how}  {auth:28} {c.description}")

Out

3 · One connector's manifest

Each connector is one declarative manifest: where it lives, how it authenticates, and the scopes it asks for.

In [4]

python

linear = next(c for c in connectors if c.name == "linear")
for field in ("name", "category", "kind", "url", "auth", "uses_oauth", "oauth", "scopes", "suggest"):
    print(f"{field:11} {getattr(linear, field)!r}")

Out

name        'linear'
category    'developer'
kind        'hosted'
url         'https://mcp.linear.app/sse'
auth        'oauth'
uses_oauth  True
oauth       'linear'
scopes      ['read', 'write', 'issues:create']
suggest     ['linear', 'issue', 'ticket']

4 · Missing keys fail early, with a clear message

connect() checks a connector's requirements before anything starts.

In [5]

python

from shipit_agent import connect

for name in ("linear", "airtable"):
    try:
        connect(name)
    except Exception as error:
        print(f"{name}: {type(error).__name__}: {error}\n")

Out

linear: ValueError: Connector 'linear' is per-user OAuth — pass token=<access token>. Run its OAuth flow (OAUTH_PRESETS['linear']) and store the token per user; refresh it when token_is_expired().

airtable: ValueError: Connector 'airtable' needs env var(s): AIRTABLE_API_KEY. Pass them via env={...} or export them.

5 · A local OAuth provider

In [6]

python

import json
import secrets
import threading
from http.server import BaseHTTPRequestHandler, HTTPServer
from urllib.parse import parse_qs, urlencode, urlparse

class Provider:
    codes: dict[str, str] = {}
    refresh_tokens: set[str] = set()
    calls: Counter = Counter()

class Handler(BaseHTTPRequestHandler):
    def log_message(self, *args):  # keep the notebook quiet
        pass

    def do_GET(self):  # the authorize page: the user approves, we redirect back with a code
        q = parse_qs(urlparse(self.path).query)
        code = secrets.token_urlsafe(8)
        Provider.codes[code] = q["client_id"][0]
        Provider.calls["authorize"] += 1
        self.send_response(302)
        self.send_header("Location", f"{q['redirect_uri'][0]}?{urlencode({'code': code, 'state': q['state'][0]})}")
        self.end_headers()

    def do_POST(self):  # the token endpoint
        form = {k: v[0] for k, v in parse_qs(self.rfile.read(int(self.headers["content-length"])).decode()).items()}
        Provider.calls[form["grant_type"]] += 1
        ok = (form["grant_type"] == "authorization_code" and Provider.codes.pop(form.get("code", ""), None)) or \
             (form["grant_type"] == "refresh_token" and form.get("refresh_token") in Provider.refresh_tokens)
        if not ok:
            self.send_response(400)
            self.end_headers()
            self.wfile.write(b'{"error": "invalid_grant"}')
            return
        refresh = secrets.token_urlsafe(12)
        Provider.refresh_tokens.add(refresh)
        body = {"access_token": "at_" + secrets.token_urlsafe(12), "refresh_token": refresh,
                "token_type": "Bearer", "expires_in": 3600}
        self.send_response(200)
        self.send_header("content-type", "application/json")
        self.end_headers()
        self.wfile.write(json.dumps(body).encode())

server = HTTPServer(("127.0.0.1", 0), Handler)
threading.Thread(target=server.serve_forever, daemon=True).start()
BASE = f"http://127.0.0.1:{server.server_port}"
print("provider running at", BASE)

Out

provider running at http://127.0.0.1:61173

6 · The flow by hand, step 1: send the user to sign in

In [7]

python

from pathlib import Path

from shipit_agent import FileOAuthStateStore, OAuthClientConfig, OAuthHelper

config = OAuthClientConfig(
    client_id="demo-client", client_secret="demo-secret",
    redirect_uri="https://app.example.com/oauth/callback",
    scopes=["read", "write"],
    authorize_url=f"{BASE}/authorize", token_url=f"{BASE}/token",
)
oauth = OAuthHelper(config, state_store=FileOAuthStateStore("oauth_state.json"))

start = oauth.create_authorization_url(state_payload={"user_id": "u_812"})
print(start["url"].replace(BASE, "<provider>"))

Out

<provider>/authorize?client_id=demo-client&redirect_uri=https%3A%2F%2Fapp.example.com%2Foauth%2Fcallback&scope=read+write&response_type=code&state=AZ2tTmoqPLj76QEDb7hS9aNxciSDK80Q

7 · Step 2: the callback brings a code

In a web app the browser comes back to your redirect URI. Here we follow the provider's redirect by hand and read the code and state from it.

In [8]

python

import http.client

def sign_in(url: str) -> dict[str, str]:
    """What the browser does: open the authorize page and land on the callback."""
    parsed = urlparse(url)
    conn = http.client.HTTPConnection(parsed.hostname, parsed.port)
    conn.request("GET", f"{parsed.path}?{parsed.query}")
    location = conn.getresponse().getheader("Location")
    return {k: v[0] for k, v in parse_qs(urlparse(location).query).items()}

callback = sign_in(start["url"])
print({k: v[:6] + "…" for k, v in callback.items()})

Out

{'code': 'acH2La…', 'state': 'AZ2tTm…'}

8 · Step 3: exchange the code for tokens

Passing state checks it against the saved nonce: the defence against login CSRF. The result is stamped with expires_at.

In [9]

python

tokens = oauth.exchange_code(code=callback["code"], state=callback["state"])
print({k: (v[:9] + "…" if isinstance(v, str) and len(v) > 12 else v) for k, v in tokens.items()})

Out

{'access_token': 'at_EczvJX…', 'refresh_token': 'E0uvWaGGn…', 'token_type': 'Bearer', 'expires_in': 3600, 'expires_at': 1794145995}

9 · A state works once

The nonce is consumed on success, so a replayed callback is refused before any request is made.

In [10]

python

try:
    oauth.exchange_code(code=callback["code"], state=callback["state"])
except ValueError as error:
    print("refused:", error)

Out

refused: Invalid or expired OAuth state — possible CSRF or replay.

10 · Refresh

Access tokens are short-lived; the refresh token gets a new one. token_is_expired says when.

In [11]

python

from shipit_agent.integrations.oauth import token_is_expired

print("expired now?", token_is_expired(tokens))
renewed = oauth.refresh_token(tokens["refresh_token"])
print("new access token differs:", renewed["access_token"] != tokens["access_token"])
print("provider calls so far:", dict(Provider.calls))

Out

expired now? False
new access token differs: True
provider calls so far: {'authorize': 1, 'authorization_code': 1, 'refresh_token': 1}

11 · `OAuthCredentialManager`: the whole flow, per user

In an app you don't juggle tokens yourself. The manager keeps one token per (user, connector) in a store, and refreshes it when it's near expiry. Register the connector's OAuth helper once.

In [12]

python

from shipit_agent.connectors import FileTokenStore, OAuthCredentialManager

manager = OAuthCredentialManager(FileTokenStore("tokens"))
manager.register("linear", OAuthHelper(config, state_store=FileOAuthStateStore("oauth_state.json")))

url = manager.authorize_url(user="ada", connector="linear")       # your "Connect Linear" button
back = sign_in(url)                                                  # Ada approves
manager.complete(connector="linear", code=back["code"], state=back["state"])   # your callback route

print("status:", manager.status(user="ada", connector="linear"))
print("ada's connections:", manager.connections(user="ada"))
print("bob's connections:", manager.connections(user="bob"))

Out

status: connected
ada's connections: {'linear': 'connected'}
bob's connections: {}

12 · Tokens that refresh themselves

access_token() returns a valid token, refreshing it first if it's expired. Here Ada's stored token is aged past its expiry, and the next call refreshes it.

In [13]

python

import time

first = manager.access_token(user="ada", connector="linear")
stored = manager.store.get("ada", "linear")
manager.store.set("ada", "linear", {**stored, "expires_at": time.time() - 10})

before = Provider.calls["refresh_token"]
second = manager.access_token(user="ada", connector="linear")
print("refreshed:", second != first, "| refresh calls:", Provider.calls["refresh_token"] - before)
print("on disk:", sorted(p.name for p in Path("tokens").rglob("*") if p.is_file()))

Out

refreshed: True | refresh calls: 1
on disk: ['linear.json']

13 · A user's token inside a tool

Your own tools can act as the signed-in user: resolve their token at call time, never store it in the prompt. The agent's session_id carries who it's working for.

In [14]

python

from shipit_agent.tools import FunctionTool

def my_issues(cycle: str = "current") -> str:
    """List the signed-in user's Linear issues for a cycle."""
    token = manager.access_token(user="ada", connector="linear")
    # A real tool would call Linear's API with: Authorization: Bearer <token>
    return json.dumps({"cycle": cycle, "auth": token[:6] + "…", "issues": [
        {"id": "ENG-412", "title": "Retry webhooks on 5xx", "state": "In review"},
        {"id": "ENG-418", "title": "Rotate the signing key", "state": "Todo"},
    ]})

agent = Agent(llm=haiku, prompt="You help Ada with her Linear work. Be brief.",
              tools=[FunctionTool.from_callable(my_issues, read_only=True)])
print(agent.run("What's on my plate this cycle, and what's closest to done?").output)

Out

You have two issues this cycle:

- **ENG-412, Retry webhooks on 5xx**: In review. This is closest to done.
- **ENG-418, Rotate the signing key**: Todo. Not started.

14 · When a token is revoked

If the user revokes access at the provider, refresh fails and the manager raises NeedsReconnect: your app shows "Reconnect Linear" instead of making calls that fail silently.

In [15]

python

Provider.refresh_tokens.clear()            # the user revoked access at the provider
stored = manager.store.get("ada", "linear")
manager.store.set("ada", "linear", {**stored, "expires_at": time.time() - 10})
try:
    manager.access_token(user="ada", connector="linear")
except Exception as error:
    print(type(error).__name__, "->", error)

Out

NeedsReconnect -> Refreshing 'linear' for ada failed: HTTP Error 400: Bad Request

15 · The same tool on Gemma 4

After Ada reconnects, Gemma 4 31B (Bedrock) runs the same tool.

In [16]

python

back = sign_in(manager.authorize_url(user="ada", connector="linear"))
manager.complete(connector="linear", code=back["code"], state=back["state"])

agent = Agent(llm=gemma, prompt="You help Ada with her Linear work. Be brief.",
              tools=[FunctionTool.from_callable(my_issues, read_only=True)])
print(agent.run("What's on my plate this cycle?").output)

Out

You have two issues this cycle:
- **ENG-412**: Retry webhooks on 5xx (In review)
- **ENG-418**: Rotate the signing key (Todo)

16 · In a web app

The same three calls map onto two routes and a per-request agent. With FastAPI:

@app.get("/connect/linear")
def connect_linear(user=Depends(current_user)):
    return RedirectResponse(manager.authorize_url(user=user.id, connector="linear"))

@app.get("/oauth/callback")
def callback(code: str, state: str):
    manager.complete(connector="linear", code=code, state=state)
    return RedirectResponse("/settings/connections")

@app.post("/ask")
def ask(body: Question, user=Depends(current_user)):
    agent = Agent(llm=llm, mcps=[manager.connect("linear", user=user.id)])
    return {"answer": agent.run(body.text).output}

manager.connect(connector, user=...) returns the hosted MCP server with that user's token attached.

17 · Clean up

In [17]

python

import shutil

server.shutdown()
shutil.rmtree("tokens", ignore_errors=True)
Path("oauth_state.json").unlink(missing_ok=True)

18 · Summary

list_connectors() is the catalog; each connector says where it lives and how it authenticates.
connect() fails early and clearly when a key or token is missing.
OAuthHelper is the authorization-code flow; state is checked and consumed, so replays are refused.
OAuthCredentialManager keeps per-user tokens, refreshes them, and raises NeedsReconnect when it can't.
Tools resolve the user's token at call time; manager.connect() attaches a hosted connector per user.

Next: Guide 06 — RAG.

Connections and OAuth

1 · Models

2 · The catalog

3 · One connector's manifest

4 · Missing keys fail early, with a clear message

5 · A local OAuth provider

6 · The flow by hand, step 1: send the user to sign in

7 · Step 2: the callback brings a code

8 · Step 3: exchange the code for tokens

9 · A state works once

10 · Refresh

11 · OAuthCredentialManager: the whole flow, per user

12 · Tokens that refresh themselves

13 · A user's token inside a tool

14 · When a token is revoked

15 · The same tool on Gemma 4

16 · In a web app

17 · Clean up

18 · Summary

Connections and OAuth

1 · Models

2 · The catalog

3 · One connector's manifest

4 · Missing keys fail early, with a clear message

5 · A local OAuth provider

6 · The flow by hand, step 1: send the user to sign in

7 · Step 2: the callback brings a code

8 · Step 3: exchange the code for tokens

9 · A state works once

10 · Refresh

11 · OAuthCredentialManager: the whole flow, per user

12 · Tokens that refresh themselves

13 · A user's token inside a tool

14 · When a token is revoked

15 · The same tool on Gemma 4

16 · In a web app

17 · Clean up

18 · Summary

11 · `OAuthCredentialManager`: the whole flow, per user

11 · `OAuthCredentialManager`: the whole flow, per user