MCP Postgres Starter

Category: AI & Agents

This page is generated from the Air Pipe marketplace. Browse it live to install into your organization.

Turn your Postgres data into secure MCP tools an AI client (Claude Desktop, Claude Code, Cursor, any MCP client) can call — with a real authentication story for two situations: just you (point an AI at your own database) and your customers (each end-user gets their own tenant-scoped view through one shared endpoint). No new auth surface to learn — MCP reuses the same per-request checks as a normal Air Pipe route.

The one thing to understand about MCP auth

Air Pipe's MCP layer adds no authentication of its own. When an MCP client calls a tool, Air Pipe takes the client's Authorization: Bearer <token>, forwards it into the interface as the airpipe-jwt header, and runs the same actions a normal HTTP request would. So securing an MCP tool is exactly securing a normal Air Pipe route — you validate the token and scope your queries in the config. That's the whole model. This pack gives you two ready-made shapes for it.

Pick your tier

	Tier 1 — Single token	Tier 2 — Per-user (multi-tenant)
For	Pointing an AI at your own database (solo dev, internal tools, trusted team)	Exposing an MCP to your customers, each seeing only their own data
Token	One shared HS256 token you mint once	One HS256 token per end-user, carrying a `tenant_id` claim
Scoping	None — the token sees everything	Every query filtered by the token's `tenant_id`
Revocation	Rotate `SOLO_SECRET`	Per-token denylist (`POST /auth/revoke`), effective on the next call
File	`tasks-solo.yml`	`tasks.yml` + `auth.yml`

Deploy one tier for real use. Both ship here (with distinct routes/tool names) so you can try them side by side; delete the one you don't want.

What's included

File	Purpose
`schema.sql`	Tables: `mcp_tenants`, `mcp_tokens` (revocation denylist), `mcp_tasks` (your data)
`seed.yml`	`POST /api/seed` — creates tables and loads two demo tenants + sample tasks
`tasks.yml`	Tier 2 tenant-scoped MCP tools: `list_tasks`, `get_task`, `create_task`, `update_task`
`auth.yml`	Tier 2 token lifecycle (HTTP, not MCP tools): mint / revoke / list / whoami
`auth-oidc.yml`	Tier 2 (OIDC) verify your IdP's RS256 tokens directly via JWKS — no exchange shim (engine ≥ 0.196.0)
`tasks-solo.yml`	Tier 1 single-token MCP tools: `list_tasks_solo`, `create_task_solo`

MCP tools

Tier 2 (`tasks.yml`)

Tool	Input	Description
`list_tasks`	`{ status? }`	List the caller's tasks (newest first), optional `open`/`done` filter
`get_task`	`{ task_id }`	Fetch one task by id
`create_task`	`{ title, status? }`	Create a task
`update_task`	`{ task_id, title?, status? }`	Update a task (omitted fields unchanged)

Every call: validate the JWT from the header → re-check the revocation denylist → validate inputs → run a query scoped to a|ValidateJwt::tenant_id|. The token never appears in a tool's input schema — it's supplied out-of-band as the connection bearer.

Tier 2 — OIDC (`auth-oidc.yml`)

Tool	Input	Description
`list_tasks_oidc`	`{ status? }`	Same as `list_tasks`, but verifies your IdP's RS256 token via JWKS and scopes to its `tenant_id` claim

A drop-in for teams whose users log in through Auth0/Clerk/Cognito — no exchange step. Copy the ValidateJwt block onto the other tools to make the whole surface OIDC-verified.

Tier 1 (`tasks-solo.yml`)

Tool	Input	Description
`list_tasks_solo`	`{ status? }`	List all tasks (single token = full access)
`create_task_solo`	`{ title, status? }`	Create a task under `DEFAULT_TENANT`

Auth lifecycle endpoints (Tier 2, `auth.yml`)

These are ordinary HTTP routes your backend calls — not MCP tools.

Method	Path	Auth	Description
`POST`	`/auth/exchange`	`x-exchange-secret`	Mint an MCP token for `{ tenant_id, subject }`
`POST`	`/auth/revoke`	`x-exchange-secret`	Revoke a token by `{ jti }`
`POST`	`/auth/tokens/list`	`x-exchange-secret`	List issued token metadata (never the token)
`GET`	`/auth/whoami`	`airpipe-jwt`	Decode & return the caller's token claims

Setup

1. Managed variables

In the Air Pipe dashboard (or as ap_vars self-hosted), add:

Name	Used by	Value
`DATABASE_URL`	all	`postgresql://user:pass@host:5432/dbname`
`JWT_SECRET`	Tier 2	32+ char HS256 secret (signs & verifies MCP tokens)
`EXCHANGE_SECRET`	Tier 2	shared secret your backend sends as `x-exchange-secret`
`SOLO_SECRET`	Tier 1	32+ char HS256 secret for your single token
`OIDC_JWKS_URL`	Tier 2 (OIDC)	your provider's JWKS, e.g. `https://YOUR.us.auth0.com/.well-known/jwks.json`
`OIDC_ISSUER`	Tier 2 (OIDC)	expected `iss` claim
`OIDC_AUDIENCE`	Tier 2 (OIDC)	expected `aud` claim (your API identifier)

2. Create the tables

Either run schema.sql directly (psql "$DATABASE_URL" -f schema.sql), or deploy seed.yml and call it:

curl -X POST https://your-airpipe-host/api/seed
# → tenants: 2, and the two demo tenant UUIDs

3. Deploy the configs for your chosen tier

Tier 1: schema.sql/seed.yml + tasks-solo.yml
Tier 2: schema.sql/seed.yml + auth.yml + tasks.yml

Connecting an MCP client

Point your client at https://your-airpipe-host/mcp (self-hosted) or https://your-airpipe-host/<org>/<env>/mcp (managed), with your token as the bearer. Example Claude Desktop / MCP client config:

{
  "mcpServers": {
    "my-tasks": {
      "url": "https://your-airpipe-host/mcp",
      "headers": { "Authorization": "Bearer <your-mcp-token>" }
    }
  }
}

For Tier 1, <your-mcp-token> is a token you mint once at jwt.io (HS256, secret = SOLO_SECRET, payload e.g. {"sub":"me","exp":9999999999}). For Tier 2, it's a token from POST /auth/exchange (below).

Quick start (Tier 2 walkthrough)

Air Pipe wraps HTTP responses in a standard {"data":{ "<Action>": { "data": ... } }} action trace, so the examples below extract the useful field with jq.

BASE=https://your-airpipe-host
EXCHANGE_SECRET=... ; ACME=11111111-1111-1111-1111-111111111111

# 1. Seed demo data (returns the two tenant UUIDs)
curl -sX POST $BASE/api/seed | jq '.data.SeedSummary.data'

# 2. Your backend mints a token for one of YOUR users, scoped to their tenant
TOKEN=$(curl -sX POST $BASE/auth/exchange \
  -H "x-exchange-secret: $EXCHANGE_SECRET" -H 'content-type: application/json' \
  -d "{\"tenant_id\":\"$ACME\",\"subject\":\"user-123\",\"name\":\"laptop\"}" \
  | jq -r '.data.Result.data.token')

# 3. That token drives the MCP tools — scoped to the user's tenant automatically
curl -sX POST $BASE/mcp -H "Authorization: Bearer $TOKEN" -H 'content-type: application/json' \
  -d '{"jsonrpc":"2.0","id":1,"method":"tools/call","params":{"name":"list_tasks","arguments":{}}}'

# 4. Revoke it — the next tool call fails immediately
JTI=...   # returned alongside the token in step 2
curl -sX POST $BASE/auth/revoke \
  -H "x-exchange-secret: $EXCHANGE_SECRET" -H 'content-type: application/json' \
  -d "{\"jti\":\"$JTI\"}"

Auth, in depth

Tier 1 — "for me": one token, full access

The single token is the grant. Any HS256 JWT signed with SOLO_SECRET passes; there's no per-row scoping, so every tool sees every row. Perfect for pointing Claude at your own database. To revoke everything, rotate SOLO_SECRET.

Tier 2 — "for my customers": per-user tokens, tenant-scoped

Your app already knows who the user is (your session, or an IdP like Auth0/Clerk/Cognito). Your backend — which trusts that login — calls POST /auth/exchange with the user's tenant_id and subject. Air Pipe mints a short-lived HS256 token carrying those claims and records its jti in the mcp_tokens denylist. You hand that token to the user's MCP client. From then on, every tool scopes its queries to the token's tenant_id — a row from another tenant simply doesn't match, so cross-tenant access is structurally impossible, not merely forbidden.

Using a real IdP (Auth0/Clerk/Cognito) — two options

Providers like Auth0, Clerk, and Cognito issue RS256 tokens signed with keys published at a JWKS URL. You have two ways to accept them:

Verify directly via JWKS (recommended, auth-oidc.yml). On engine ≥ 0.196.0, point is_valid_jwt at the provider's JWKS — Air Pipe fetches and caches the keys, selects the signer by the token's kid, and enforces iss/aud/exp. No exchange hop, and provider key rotation just works. Configure your IdP to include a tenant_id claim and you're done:
```
is_valid_jwt:
  jwks_url: a|ap_var::OIDC_JWKS_URL|
  alg: RS256
  iss: a|ap_var::OIDC_ISSUER|
  aud: a|ap_var::OIDC_AUDIENCE|
```
Exchange to an HS256 token (auth.yml). If you'd rather not expose the JWKS path, your backend (which already trusts its provider) calls POST /auth/exchange to re-mint a short-lived HS256 token. This also gives you the jti revocation denylist.

If you already issue your own HS256 tokens with a tenant_id claim, skip both — the tasks.yml tools accept any valid HS256 token; the denylist check is gracefully skipped for tokens minted without a jti.

Revocation

JWTs are stateless — a signature check alone can't tell a revoked token from a valid one. So every Tier-2 tool re-checks the token's jti against the mcp_tokens table (revoked_at IS NULL AND expires_at > now()). POST /auth/revoke flips revoked_at; the very next tool call returns 401. This is the piece a naive JWT setup forgets.

Customisation

Swap mcp_tasks for your own table(s). The only rule for Tier 2 is that each tenant-owned table carries a tenant_id you filter on. To expose a new tool: copy an interface in tasks.yml, keep the ValidateJwt + CheckTokenActive prefix, point CheckBody/the query at your columns, and set a unique mcp.tool_name.

Notes & limitations

Algorithms. is_valid_jwt supports HS256/384/512 (shared secret), and — on engine ≥ 0.196.0 — RS256/PS256/ES256/EdDSA via a static public_key or a provider jwks_url (see auth-oidc.yml). On older engines, only HS256 is available; use the exchange shim.
Tokens are long-lived bearers. MCP clients today authenticate with a static bearer pasted into config; there's no interactive OAuth login flow yet. Keep exp reasonable and rely on the denylist (self-issued) or short IdP lifetimes (OIDC) for revocation.
Multi-statement DDL. seed.yml runs its whole schema block in one multi: true action (engine ≥ 0.196.0). On older engines, split DDL into one statement per action instead.
Response shape. Self-hosted responses are wrapped in the {"data":{...}} action trace shown above; MCP clients parse the tool result for you.

Configuration

auth-oidc.yml

name: McpPgTasksOidc
description: Tier 2 variant that verifies your identity provider's RS256 tokens DIRECTLY via its JWKS (Auth0, Clerk, Cognito, Firebase) — no exchange shim. The engine fetches and caches the provider's public keys and selects the signer by the token's key id, so provider key rotation just works.

docs: true

# TIER 2 (OIDC) — verify real IdP tokens directly.
#
# If your end-users log in through a standard OIDC provider (Auth0, Clerk,
# Cognito, Firebase, Okta…), those tokens are RS256 and signed with keys the
# provider publishes at a JWKS URL. Point `is_valid_jwt` at that JWKS and the
# engine verifies the token itself — no /auth/exchange hop, no re-minting.
#
# Requires engine >= 0.196.0 (RS256/ES256 + JWKS support in is_valid_jwt).
#
# HOW THE TENANT IS SCOPED
#   Configure your IdP to include a `tenant_id` claim in the access token
#   (Auth0 Action, Clerk JWT template, Cognito pre-token-generation trigger,
#   etc.). This tool reads that claim and scopes every query to it — exactly
#   like tasks.yml, but the trust root is your IdP's signature instead of a
#   shared HS256 secret.
#
# REVOCATION is the IdP's job here (short token lifetimes / the provider's
# revocation), so there's no denylist step — unlike the self-issued tokens in
# tasks.yml. Use short `exp`s at the IdP.
#
# Required managed variables:
#   DATABASE_URL    — Postgres connection string
#   OIDC_JWKS_URL   — your provider's JWKS, e.g. https://YOUR.us.auth0.com/.well-known/jwks.json
#   OIDC_ISSUER     — expected `iss`, e.g. https://YOUR.us.auth0.com/
#   OIDC_AUDIENCE   — expected `aud` (your API identifier)

global:
  databases:
    main:
      driver: postgres
      conn_string: "a|ap_var::DATABASE_URL|"

interfaces:

  # MCP tool: list_tasks_oidc
  oidc/tasks:
    output: http
    method: POST
    summary: List tasks (OIDC)
    description: List the caller's tasks, verifying the caller's IdP RS256 token via JWKS and scoping to its tenant_id claim.
    tags: [tasks, oidc]
    mcp:
      enabled: true
      tool_name: list_tasks_oidc
      description: List your tasks, newest first. Optional status filter ("open" or "done").
    request_example:
      status: open
    response_example:
      - id: 3f9a...
        title: Ship the MCP launch post
        status: open

    actions:
      - name: ValidateJwt
        input: a|headers|
        hide_data_on_success: true
        assert:
          http_code_on_error: 401
          error_message: "Invalid or missing token"
          tests:
            - value: airpipe-jwt
              is_not_null: true
              # Verify against the provider's rotating JWKS. The signing key is
              # chosen by the token's `kid`; iss/aud are enforced too.
              is_valid_jwt:
                jwks_url: a|ap_var::OIDC_JWKS_URL|
                alg: RS256
                iss: a|ap_var::OIDC_ISSUER|
                aud: a|ap_var::OIDC_AUDIENCE|
        post_transforms:
          - extract_value: jwt_claims

      - name: CheckBody
        run_when_succeeded:
          actions: [ValidateJwt]
          http_code_on_error: 400
        input: a|body|
        hide_data_on_success: true
        assert:
          tests:
            - value: status
              is_not_null: false
              description: Optional status filter — "open" or "done".

      - name: ListTasks
        run_when_succeeded: [CheckBody]
        database: main
        # tenant_id comes from the IdP token claim, bound as a parameter.
        query: |
          SELECT id, title, status, created_at, updated_at
          FROM mcp_tasks
          WHERE tenant_id = $1::uuid
            AND ($2::text IS NULL OR status = $2::text)
          ORDER BY created_at DESC
          LIMIT 200;
        params:
          - a|ValidateJwt::tenant_id|
          - a|body::status->default(null)|

auth.yml

name: McpPgAuth
description: Tier 2 token lifecycle for the MCP Postgres Starter — mint (exchange), revoke, list, and whoami. These are ordinary HTTP routes your backend calls; they are NOT exposed as MCP tools.

docs: true

# Token lifecycle for Tier 2 (per-user, multi-tenant MCP).
#
# THE FLOW
#   1. Your app authenticates the end-user however you already do (session,
#      Auth0/Clerk/Cognito, your own login). YOU trust that result.
#   2. Your backend calls POST /auth/exchange with a shared EXCHANGE_SECRET and
#      the user's { tenant_id, subject }. Air Pipe mints a short-lived HS256 MCP
#      token and records its jti in the mcp_tokens denylist table.
#   3. You hand that token to the user; they paste it into their MCP client as
#      the bearer credential. Every MCP tool call is then scoped to their tenant.
#   4. Revoke any token with POST /auth/revoke (flips revoked_at); the MCP tools
#      re-check the denylist on every call, so revocation takes effect instantly.
#
# WHY AN EXCHANGE STEP?  Air Pipe's is_valid_jwt verifies HS256 (shared-secret)
# tokens only — it cannot verify the RS256/JWKS tokens a provider like Auth0
# issues. So your backend (which already trusts its provider) re-mints an HS256
# token here. If you already issue your own HS256 tokens with a tenant_id claim,
# you can skip exchange entirely and point clients straight at the MCP tools.
#
# Required managed variables:
#   DATABASE_URL    — Postgres connection string (pgcrypto enabled)
#   JWT_SECRET      — 32+ char HS256 secret used to sign & verify MCP tokens
#   EXCHANGE_SECRET — shared secret your backend sends as the x-exchange-secret
#                     header to authorize minting/revoking tokens

global:
  variables:
    TOKEN_TTL: "90d"
  databases:
    main:
      driver: postgres
      conn_string: "a|ap_var::DATABASE_URL|"

interfaces:

  # POST /auth/exchange
  # Header: x-exchange-secret: <EXCHANGE_SECRET>
  # Body:   { "tenant_id": "<uuid>", "subject": "user-123", "name": "laptop" }
  # Returns a fresh MCP token (shown once) + its jti and expiry.
  auth/exchange:
    output: http
    method: POST
    summary: Mint an MCP token
    description: Exchange a trusted (tenant_id, subject) for a short-lived HS256 MCP token. Authorized by the x-exchange-secret header.
    tags: [auth]
    request_example:
      tenant_id: 11111111-1111-1111-1111-111111111111
      subject: user-123
      name: laptop
    response_example:
      jti: 9b2c...
      expires_at: "2026-10-01T12:00:00Z"
      token: eyJhbGciOiJIUzI1Ni...

    actions:
      # Gate: only a caller holding the shared EXCHANGE_SECRET may mint tokens.
      - name: Authorize
        input: a|headers|
        hide_data_on_success: true
        assert:
          http_code_on_error: 401
          error_message: "Missing or invalid x-exchange-secret"
          tests:
            - value: x-exchange-secret
              is_not_null: true
              is_equal_to: a|ap_var::EXCHANGE_SECRET|

      - name: Validate
        run_when_succeeded:
          actions: [Authorize]
          http_code_on_error: 401
        input: a|body|
        hide_data_on_success: true
        assert:
          http_code_on_error: 400
          error_message: "tenant_id (uuid) and subject are required"
          tests:
            - value: tenant_id
              is_not_null: true
              is_uuid: true
            - value: subject
              is_not_null: true
              is_not_empty: true

      # Mint: attach a random jti, then sign an HS256 token carrying the tenant
      # scope. add_jwt adds exp (from TOKEN_TTL) and iss automatically.
      - name: Mint
        run_when_succeeded:
          actions: [Validate]
          http_code_on_error: 400
        input: a|Validate|
        hide_data_on_success: true
        post_transforms:
          - add_attribute:
              jti: a|uuid|
          - add_jwt:
              key: token
              secret: a|ap_var::JWT_SECRET|
              exp: a|var::TOKEN_TTL|
              data: [tenant_id, subject, jti]

      # Record the jti in the denylist table so it can be revoked / expired.
      - name: Record
        run_when_succeeded:
          actions: [Mint]
          http_code_on_error: 500
        database: main
        hide_data_on_success: true
        query: |
          INSERT INTO mcp_tokens (jti, tenant_id, subject, name, expires_at)
          VALUES ($1::uuid, $2::uuid, $3, COALESCE($4, 'default'), NOW() + INTERVAL '90 days')
          RETURNING jti, expires_at;
        params:
          - a|Mint::jti|
          - a|Validate::tenant_id|
          - a|Validate::subject|
          - a|body::name->default(null)|
        post_transforms:
          - extract_value: "[0]"

      - name: Result
        run_when_succeeded: [Record]
        input: a|Record|
        post_transforms:
          - add_attribute:
              token: a|Mint::token|

  # GET /auth/whoami
  # Header: airpipe-jwt: <mcp token>   (or Authorization: Bearer <mcp token>)
  # Decodes and returns the token's claims — a quick "is my token valid?" check.
  auth/whoami:
    output: http
    summary: Inspect the current MCP token
    description: Validate the caller's MCP token and return its claims (tenant_id, subject, jti, exp).
    tags: [auth]
    response_example:
      tenant_id: 11111111-1111-1111-1111-111111111111
      subject: user-123
      jti: 9b2c...
      exp: 1790000000

    actions:
      - name: ValidateToken
        input: a|headers|
        assert:
          http_code_on_error: 401
          error_message: "Invalid or missing token"
          tests:
            - value: airpipe-jwt
              is_not_null: true
              is_valid_jwt: a|ap_var::JWT_SECRET|
        post_transforms:
          - extract_value: jwt_claims

  # POST /auth/revoke
  # Header: x-exchange-secret: <EXCHANGE_SECRET>   Body: { "jti": "<uuid>" }
  # Revokes a previously issued token. Takes effect on the next MCP tool call.
  auth/revoke:
    output: http
    method: POST
    summary: Revoke an MCP token
    description: Revoke a token by its jti. The MCP tools re-check the denylist on every call, so revocation is immediate.
    tags: [auth]
    request_example:
      jti: 9b2c...
    response_example:
      jti: 9b2c...
      revoked_at: "2026-07-03T09:00:00Z"

    actions:
      - name: Authorize
        input: a|headers|
        hide_data_on_success: true
        assert:
          http_code_on_error: 401
          error_message: "Missing or invalid x-exchange-secret"
          tests:
            - value: x-exchange-secret
              is_not_null: true
              is_equal_to: a|ap_var::EXCHANGE_SECRET|

      - name: Validate
        run_when_succeeded:
          actions: [Authorize]
          http_code_on_error: 401
        input: a|body|
        hide_data_on_success: true
        assert:
          http_code_on_error: 400
          error_message: "jti (uuid) is required"
          tests:
            - value: jti
              is_not_null: true
              is_uuid: true

      - name: Revoke
        run_when_succeeded:
          actions: [Validate]
          http_code_on_error: 400
        database: main
        query: |
          UPDATE mcp_tokens
          SET revoked_at = NOW()
          WHERE jti = $1::uuid AND revoked_at IS NULL
          RETURNING jti, revoked_at;
        params:
          - a|Validate::jti|
        assert:
          http_code_on_error: 404
          error_message: "Token not found or already revoked"
          tests:
            - value: count()
              is_equal_to: 1
        post_transforms:
          - extract_value: "[0]"

  # POST /auth/tokens/list
  # Header: x-exchange-secret: <EXCHANGE_SECRET>
  # Body (optional): { "tenant_id": "<uuid>" } to filter to one tenant.
  # Lists issued tokens (never the token string — only metadata).
  auth/tokens/list:
    output: http
    method: POST
    summary: List issued MCP tokens
    description: List token metadata (jti, subject, revoked, expiry). Optionally filter by tenant_id. Never returns the token string.
    tags: [auth]
    response_example:
      - jti: 9b2c...
        subject: user-123
        name: laptop
        revoked: false
        expires_at: "2026-10-01T12:00:00Z"

    actions:
      - name: Authorize
        input: a|headers|
        hide_data_on_success: true
        assert:
          http_code_on_error: 401
          error_message: "Missing or invalid x-exchange-secret"
          tests:
            - value: x-exchange-secret
              is_not_null: true
              is_equal_to: a|ap_var::EXCHANGE_SECRET|

      - name: ListTokens
        run_when_succeeded:
          actions: [Authorize]
          http_code_on_error: 401
        database: main
        query: |
          SELECT jti, subject, name,
                 (revoked_at IS NOT NULL) AS revoked,
                 expires_at, created_at
          FROM mcp_tokens
          WHERE $1::uuid IS NULL OR tenant_id = $1::uuid
          ORDER BY created_at DESC;
        params:
          - a|body::tenant_id->default(null)|

seed.yml

name: McpPgSeed
description: Creates all tables (idempotent) and loads two demo tenants with fixed UUIDs plus sample tasks. Safe to re-run — clears data first. Returns the tenant UUIDs so you can mint MCP tokens immediately.

docs: true

# Required managed variable:
#   DATABASE_URL — Postgres connection string (pgcrypto enabled)

global:
  databases:
    main:
      driver: postgres
      conn_string: "a|ap_var::DATABASE_URL|"

interfaces:

  # POST /api/seed
  # Creates the schema and loads demo data. Returns the two demo tenant UUIDs so
  # you can mint a token (POST /auth/exchange) and start calling the MCP tools.
  api/seed:
    output: http
    method: POST
    summary: Seed demo schema and data
    description: Idempotently creates the tables and loads two demo tenants (Acme, Globex) with sample tasks.
    tags: [setup]
    response_example:
      tenants: 2
      tasks: 3
      demo_tenants:
        Acme Inc: 11111111-1111-1111-1111-111111111111
        Globex Corp: 22222222-2222-2222-2222-222222222222

    actions:
      # `multi: true` runs a whole multi-statement DDL block in one action. Without
      # it the native Postgres driver prepares the query, and a prepared statement
      # can hold only one command (SQLSTATE 42601). Requires engine >= 0.196.0.
      - name: CreateSchema
        database: main
        multi: true
        hide_data_on_success: true
        query: |
          CREATE EXTENSION IF NOT EXISTS pgcrypto;

          CREATE TABLE IF NOT EXISTS mcp_tenants (
            id         UUID        PRIMARY KEY DEFAULT gen_random_uuid(),
            name       TEXT        NOT NULL,
            created_at TIMESTAMPTZ NOT NULL DEFAULT NOW()
          );

          CREATE TABLE IF NOT EXISTS mcp_tokens (
            jti        UUID        PRIMARY KEY,
            tenant_id  UUID        NOT NULL REFERENCES mcp_tenants(id) ON DELETE CASCADE,
            subject    TEXT        NOT NULL,
            name       TEXT        NOT NULL DEFAULT 'default',
            created_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
            expires_at TIMESTAMPTZ NOT NULL,
            revoked_at TIMESTAMPTZ
          );

          CREATE TABLE IF NOT EXISTS mcp_tasks (
            id         UUID        PRIMARY KEY DEFAULT gen_random_uuid(),
            tenant_id  UUID        NOT NULL REFERENCES mcp_tenants(id) ON DELETE CASCADE,
            title      TEXT        NOT NULL,
            status     TEXT        NOT NULL DEFAULT 'open' CHECK (status IN ('open', 'done')),
            created_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
            updated_at TIMESTAMPTZ NOT NULL DEFAULT NOW()
          );

          CREATE INDEX IF NOT EXISTS idx_mcp_tokens_tenant ON mcp_tokens (tenant_id);
          CREATE INDEX IF NOT EXISTS idx_mcp_tasks_tenant  ON mcp_tasks  (tenant_id, created_at DESC);

      - name: ClearTables
        run_when_succeeded: [CreateSchema]
        database: main
        hide_data_on_success: true
        query: |
          TRUNCATE mcp_tokens, mcp_tasks, mcp_tenants RESTART IDENTITY CASCADE;

      # Fixed UUIDs so the demo is reproducible: mint a token for either id below.
      - name: InsertTenants
        run_when_succeeded: [ClearTables]
        database: main
        hide_data_on_success: true
        query: |
          INSERT INTO mcp_tenants (id, name) VALUES
            ('11111111-1111-1111-1111-111111111111', 'Acme Inc'),
            ('22222222-2222-2222-2222-222222222222', 'Globex Corp');

      - name: InsertTasks
        run_when_succeeded: [InsertTenants]
        database: main
        hide_data_on_success: true
        query: |
          INSERT INTO mcp_tasks (tenant_id, title, status)
          SELECT v.tenant_id::uuid, v.title, v.status
          FROM (VALUES
            ('11111111-1111-1111-1111-111111111111', 'Ship the MCP launch post', 'open'),
            ('11111111-1111-1111-1111-111111111111', 'Review Q3 numbers',        'done'),
            ('22222222-2222-2222-2222-222222222222', 'Migrate data pipeline',    'open')
          ) AS v(tenant_id, title, status);

      - name: SeedSummary
        run_when_succeeded: [InsertTasks]
        database: main
        query: |
          SELECT
            (SELECT COUNT(*) FROM mcp_tenants) AS tenants,
            (SELECT COUNT(*) FROM mcp_tasks)   AS tasks,
            json_build_object(
              'Acme Inc',    '11111111-1111-1111-1111-111111111111',
              'Globex Corp', '22222222-2222-2222-2222-222222222222'
            ) AS demo_tenants;
        post_transforms:
          - extract_value: "[0]"

tasks-solo.yml

name: McpPgTasksSolo
description: Tier 1 (single-token) MCP tools over Postgres. One shared HS256 token grants full access — the simplest way to point Claude Desktop/Code at your own database. No per-user isolation; use Tier 2 (tasks.yml) when your customers each need their own scoped view.

docs: true

# TIER 1 — one token, full access. Best for pointing an AI client at YOUR OWN
# database (internal tools, a solo dev, a trusted team).
#
# There is no tenant scoping here: the single token is the whole grant, so every
# tool sees every row. Auth is one step — validate an HS256 token from the
# airpipe-jwt header (the MCP client sends it as `Authorization: Bearer`).
#
# GET A TOKEN (once): go to https://jwt.io, algorithm HS256, set the secret to
# your SOLO_SECRET, payload e.g. { "sub": "me", "exp": <far-future> }, copy the
# token, and paste it into your MCP client. Rotate SOLO_SECRET to revoke.
#
# These tools use `solo/` routes and `_solo` tool names so they can sit alongside
# Tier 2 while you evaluate. In production, keep only ONE tier and delete the other.
#
# Required managed variables:
#   DATABASE_URL — Postgres connection string
#   SOLO_SECRET  — 32+ char HS256 secret for your single access token

global:
  variables:
    # The tenant new tasks are filed under (Tier 1 ignores tenancy otherwise).
    DEFAULT_TENANT: "11111111-1111-1111-1111-111111111111"
  databases:
    main:
      driver: postgres
      conn_string: "a|ap_var::DATABASE_URL|"

interfaces:

  # MCP tool: list_tasks_solo
  solo/tasks:
    output: http
    method: POST
    summary: List all tasks (single-token)
    description: List every task across the database. Guarded by the single shared token.
    tags: [tasks, solo]
    mcp:
      enabled: true
      tool_name: list_tasks_solo
      description: List all tasks. Optional status filter ("open" or "done").
    request_example:
      status: open
    response_example:
      - id: 3f9a...
        title: Ship the MCP launch post
        status: open

    actions:
      - name: ValidateToken
        input: a|headers|
        hide_data_on_success: true
        assert:
          http_code_on_error: 401
          error_message: "Invalid or missing token"
          tests:
            - value: airpipe-jwt
              is_not_null: true
              is_valid_jwt: a|ap_var::SOLO_SECRET|
        post_transforms:
          - extract_value: jwt_claims

      - name: CheckBody
        run_when_succeeded:
          actions: [ValidateToken]
          http_code_on_error: 400
        input: a|body|
        hide_data_on_success: true
        assert:
          tests:
            - value: status
              is_not_null: false
              description: Optional status filter — "open" or "done".

      - name: ListTasks
        run_when_succeeded: [CheckBody]
        database: main
        query: |
          SELECT id, tenant_id, title, status, created_at, updated_at
          FROM mcp_tasks
          WHERE ($1::text IS NULL OR status = $1::text)
          ORDER BY created_at DESC
          LIMIT 200;
        params:
          - a|body::status->default(null)|

  # MCP tool: create_task_solo
  solo/tasks/create:
    output: http
    method: POST
    summary: Create a task (single-token)
    description: Create a task. Files it under DEFAULT_TENANT (Tier 1 ignores tenancy).
    tags: [tasks, solo]
    mcp:
      enabled: true
      tool_name: create_task_solo
      description: Create a new task. Requires a title; status defaults to "open".
    request_example:
      title: Draft the changelog
      status: open
    response_example:
      id: 7c1e...
      title: Draft the changelog
      status: open

    actions:
      - name: ValidateToken
        input: a|headers|
        hide_data_on_success: true
        assert:
          http_code_on_error: 401
          error_message: "Invalid or missing token"
          tests:
            - value: airpipe-jwt
              is_not_null: true
              is_valid_jwt: a|ap_var::SOLO_SECRET|
        post_transforms:
          - extract_value: jwt_claims

      - name: CheckBody
        run_when_succeeded:
          actions: [ValidateToken]
          http_code_on_error: 400
        input: a|body|
        hide_data_on_success: true
        assert:
          http_code_on_error: 400
          error_message: "title is required"
          tests:
            - value: title
              is_not_null: true
              is_not_empty: true
              description: The task title.
            - value: status
              is_not_null: false
              description: Optional status — "open" (default) or "done".

      - name: CreateTask
        run_when_succeeded: [CheckBody]
        database: main
        query: |
          INSERT INTO mcp_tasks (tenant_id, title, status)
          VALUES ($1::uuid, $2, COALESCE($3, 'open'))
          RETURNING id, tenant_id, title, status, created_at, updated_at;
        params:
          - a|var::DEFAULT_TENANT|
          - a|CheckBody::title|
          - a|body::status->default(null)|
        post_transforms:
          - extract_value: "[0]"

tasks.yml

name: McpPgTasks
description: Tier 2 (multi-tenant) MCP tools over Postgres. Every tool validates a per-user HS256 token, re-checks the revocation denylist, and scopes every query to the tenant_id carried in the token — so one MCP endpoint safely serves all your customers.

docs: true

# TIER 2 — per-user, multi-tenant MCP tools.
#
# How each tool is secured, in order:
#   1. ValidateJwt      — verify the HS256 token from the airpipe-jwt header
#                         (the MCP client sends it as `Authorization: Bearer`;
#                         Air Pipe forwards it into the airpipe-jwt slot).
#   2. CheckTokenActive — re-check the token's jti against the mcp_tokens
#                         denylist so revoked/expired tokens are rejected
#                         instantly. Tokens minted without a jti skip this
#                         (they rely on the signature + exp only).
#   3. CheckBody        — validate the tool's business inputs. Only this action
#                         reads a|body|, so it — and nothing about auth — defines
#                         the MCP tool's inputSchema.
#   4. <the query>      — scoped with `a|ValidateJwt::tenant_id|` bound as a
#                         parameter, never string-interpolated. A row from
#                         another tenant simply doesn't match → no cross-tenant leak.
#
# Mint tokens with POST /auth/exchange (see auth.yml).
#
# Required managed variables:
#   DATABASE_URL — Postgres connection string
#   JWT_SECRET   — 32+ char HS256 secret (must match the one auth.yml signs with)

global:
  databases:
    main:
      driver: postgres
      conn_string: "a|ap_var::DATABASE_URL|"

interfaces:

  # MCP tool: list_tasks
  tasks:
    output: http
    method: POST
    summary: List tasks
    description: List the caller's tasks, most recent first. Optionally filter by status.
    tags: [tasks]
    mcp:
      enabled: true
      tool_name: list_tasks
      description: List your tasks, newest first. Optional status filter ("open" or "done").
    request_example:
      status: open
    response_example:
      - id: 3f9a...
        title: Ship the MCP launch post
        status: open
        created_at: "2026-07-01T12:00:00Z"

    actions:
      - name: ValidateJwt
        input: a|headers|
        hide_data_on_success: true
        assert:
          http_code_on_error: 401
          error_message: "Invalid or missing token"
          tests:
            - value: airpipe-jwt
              is_not_null: true
              is_valid_jwt: a|ap_var::JWT_SECRET|
        post_transforms:
          - extract_value: jwt_claims

      - name: CheckTokenActive
        run_when_succeeded:
          actions: [ValidateJwt]
          http_code_on_error: 401
        database: main
        hide_data_on_success: true
        query: |
          SELECT (
            $1::uuid IS NULL OR EXISTS (
              SELECT 1 FROM mcp_tokens
              WHERE jti = $1::uuid AND revoked_at IS NULL AND expires_at > NOW()
            )
          ) AS ok;
        params:
          - a|ValidateJwt::jti->default(null)|
        assert:
          http_code_on_error: 401
          error_message: "Token revoked or expired"
          tests:
            - value: "[0]ok"
              is_equal_to: true

      - name: CheckBody
        run_when_succeeded:
          actions: [CheckTokenActive]
          http_code_on_error: 400
        input: a|body|
        hide_data_on_success: true
        assert:
          tests:
            # Optional input — declared so it surfaces on the MCP tool schema.
            # is_not_null:false is an always-pass predicate that keeps it optional.
            - value: status
              is_not_null: false
              description: Optional status filter — "open" or "done".

      - name: ListTasks
        run_when_succeeded: [CheckBody]
        database: main
        query: |
          SELECT id, title, status, created_at, updated_at
          FROM mcp_tasks
          WHERE tenant_id = $1::uuid
            AND ($2::text IS NULL OR status = $2::text)
          ORDER BY created_at DESC
          LIMIT 200;
        params:
          - a|ValidateJwt::tenant_id|
          - a|body::status->default(null)|

  # MCP tool: get_task
  tasks/get:
    output: http
    method: POST
    summary: Get a task
    description: Fetch a single task by id (scoped to the caller's tenant).
    tags: [tasks]
    mcp:
      enabled: true
      tool_name: get_task
      description: Get one task by its id.
    request_example:
      task_id: 3f9a1c2d-0000-0000-0000-000000000000
    response_example:
      id: 3f9a...
      title: Ship the MCP launch post
      status: open

    actions:
      - name: ValidateJwt
        input: a|headers|
        hide_data_on_success: true
        assert:
          http_code_on_error: 401
          error_message: "Invalid or missing token"
          tests:
            - value: airpipe-jwt
              is_not_null: true
              is_valid_jwt: a|ap_var::JWT_SECRET|
        post_transforms:
          - extract_value: jwt_claims

      - name: CheckTokenActive
        run_when_succeeded:
          actions: [ValidateJwt]
          http_code_on_error: 401
        database: main
        hide_data_on_success: true
        query: |
          SELECT (
            $1::uuid IS NULL OR EXISTS (
              SELECT 1 FROM mcp_tokens
              WHERE jti = $1::uuid AND revoked_at IS NULL AND expires_at > NOW()
            )
          ) AS ok;
        params:
          - a|ValidateJwt::jti->default(null)|
        assert:
          http_code_on_error: 401
          error_message: "Token revoked or expired"
          tests:
            - value: "[0]ok"
              is_equal_to: true

      - name: CheckBody
        run_when_succeeded:
          actions: [CheckTokenActive]
          http_code_on_error: 400
        input: a|body|
        hide_data_on_success: true
        assert:
          http_code_on_error: 400
          error_message: "task_id (uuid) is required"
          tests:
            - value: task_id
              is_not_null: true
              is_uuid: true
              description: The id of the task to fetch.

      - name: GetTask
        run_when_succeeded: [CheckBody]
        database: main
        query: |
          SELECT id, title, status, created_at, updated_at
          FROM mcp_tasks
          WHERE tenant_id = $1::uuid AND id = $2::uuid;
        params:
          - a|ValidateJwt::tenant_id|
          - a|CheckBody::task_id|
        assert:
          http_code_on_error: 404
          error_message: "Task not found"
          tests:
            - value: count()
              is_equal_to: 1
        post_transforms:
          - extract_value: "[0]"

  # MCP tool: create_task
  tasks/create:
    output: http
    method: POST
    summary: Create a task
    description: Create a task for the caller's tenant.
    tags: [tasks]
    mcp:
      enabled: true
      tool_name: create_task
      description: Create a new task. Requires a title; status defaults to "open".
    request_example:
      title: Draft the changelog
      status: open
    response_example:
      id: 7c1e...
      title: Draft the changelog
      status: open

    actions:
      - name: ValidateJwt
        input: a|headers|
        hide_data_on_success: true
        assert:
          http_code_on_error: 401
          error_message: "Invalid or missing token"
          tests:
            - value: airpipe-jwt
              is_not_null: true
              is_valid_jwt: a|ap_var::JWT_SECRET|
        post_transforms:
          - extract_value: jwt_claims

      - name: CheckTokenActive
        run_when_succeeded:
          actions: [ValidateJwt]
          http_code_on_error: 401
        database: main
        hide_data_on_success: true
        query: |
          SELECT (
            $1::uuid IS NULL OR EXISTS (
              SELECT 1 FROM mcp_tokens
              WHERE jti = $1::uuid AND revoked_at IS NULL AND expires_at > NOW()
            )
          ) AS ok;
        params:
          - a|ValidateJwt::jti->default(null)|
        assert:
          http_code_on_error: 401
          error_message: "Token revoked or expired"
          tests:
            - value: "[0]ok"
              is_equal_to: true

      - name: CheckBody
        run_when_succeeded:
          actions: [CheckTokenActive]
          http_code_on_error: 400
        input: a|body|
        hide_data_on_success: true
        assert:
          http_code_on_error: 400
          error_message: "title is required"
          tests:
            - value: title
              is_not_null: true
              is_not_empty: true
              description: The task title.
            - value: status
              is_not_null: false
              description: Optional status — "open" (default) or "done".

      - name: CreateTask
        run_when_succeeded: [CheckBody]
        database: main
        query: |
          INSERT INTO mcp_tasks (tenant_id, title, status)
          VALUES ($1::uuid, $2, COALESCE($3, 'open'))
          RETURNING id, title, status, created_at, updated_at;
        params:
          - a|ValidateJwt::tenant_id|
          - a|CheckBody::title|
          - a|body::status->default(null)|
        post_transforms:
          - extract_value: "[0]"

  # MCP tool: update_task
  tasks/update:
    output: http
    method: POST
    summary: Update a task
    description: Update a task's title and/or status (scoped to the caller's tenant).
    tags: [tasks]
    mcp:
      enabled: true
      tool_name: update_task
      description: Update a task by id. Provide any of title or status; omitted fields are left unchanged.
    request_example:
      task_id: 3f9a1c2d-0000-0000-0000-000000000000
      status: done
    response_example:
      id: 3f9a...
      title: Ship the MCP launch post
      status: done

    actions:
      - name: ValidateJwt
        input: a|headers|
        hide_data_on_success: true
        assert:
          http_code_on_error: 401
          error_message: "Invalid or missing token"
          tests:
            - value: airpipe-jwt
              is_not_null: true
              is_valid_jwt: a|ap_var::JWT_SECRET|
        post_transforms:
          - extract_value: jwt_claims

      - name: CheckTokenActive
        run_when_succeeded:
          actions: [ValidateJwt]
          http_code_on_error: 401
        database: main
        hide_data_on_success: true
        query: |
          SELECT (
            $1::uuid IS NULL OR EXISTS (
              SELECT 1 FROM mcp_tokens
              WHERE jti = $1::uuid AND revoked_at IS NULL AND expires_at > NOW()
            )
          ) AS ok;
        params:
          - a|ValidateJwt::jti->default(null)|
        assert:
          http_code_on_error: 401
          error_message: "Token revoked or expired"
          tests:
            - value: "[0]ok"
              is_equal_to: true

      - name: CheckBody
        run_when_succeeded:
          actions: [CheckTokenActive]
          http_code_on_error: 400
        input: a|body|
        hide_data_on_success: true
        assert:
          http_code_on_error: 400
          error_message: "task_id (uuid) is required"
          tests:
            - value: task_id
              is_not_null: true
              is_uuid: true
              description: The id of the task to update.
            - value: title
              is_not_null: false
              description: New title (optional).
            - value: status
              is_not_null: false
              description: New status — "open" or "done" (optional).

      - name: UpdateTask
        run_when_succeeded: [CheckBody]
        database: main
        # COALESCE leaves a column unchanged when its param is null (field omitted).
        query: |
          UPDATE mcp_tasks
          SET title      = COALESCE($3, title),
              status     = COALESCE($4, status),
              updated_at = NOW()
          WHERE tenant_id = $1::uuid AND id = $2::uuid
          RETURNING id, title, status, created_at, updated_at;
        params:
          - a|ValidateJwt::tenant_id|
          - a|CheckBody::task_id|
          - a|body::title->default(null)|
          - a|body::status->default(null)|
        assert:
          http_code_on_error: 404
          error_message: "Task not found"
          tests:
            - value: count()
              is_equal_to: 1
        post_transforms:
          - extract_value: "[0]"

schema.sql

-- MCP Postgres Starter schema (Postgres)
--
-- Run once before deploying, or deploy seed.yml and call POST /api/seed.
-- pgcrypto provides gen_random_uuid() on older Postgres (built in on 13+).
CREATE EXTENSION IF NOT EXISTS pgcrypto;

-- A tenant = one of YOUR customers (a workspace / organisation / account).
-- Every row of your data is scoped to exactly one tenant. In Tier 1 (single
-- token, just you) you can ignore tenants entirely; in Tier 2 (per-user tokens)
-- the tenant_id is carried as a JWT claim and enforced on every query.
CREATE TABLE IF NOT EXISTS mcp_tenants (
  id         UUID        PRIMARY KEY DEFAULT gen_random_uuid(),
  name       TEXT        NOT NULL,
  created_at TIMESTAMPTZ NOT NULL DEFAULT NOW()
);

-- Issued MCP access tokens (Tier 2). This table is the REVOCATION denylist:
-- exchange mints a token + inserts its jti here; every MCP tool re-checks the
-- jti is still present, not revoked, and not expired. Revoke = flip revoked_at.
-- The token string itself is NEVER stored — only its jti (a random id claim).
CREATE TABLE IF NOT EXISTS mcp_tokens (
  jti        UUID        PRIMARY KEY,
  tenant_id  UUID        NOT NULL REFERENCES mcp_tenants(id) ON DELETE CASCADE,
  subject    TEXT        NOT NULL,               -- your end-user's id (from your IdP)
  name       TEXT        NOT NULL DEFAULT 'default',
  created_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
  expires_at TIMESTAMPTZ NOT NULL,
  revoked_at TIMESTAMPTZ
);

-- The example resource your MCP tools read and write. Swap this for your own
-- table(s) — the only rule for Tier 2 is that every tenant-owned table carries
-- a tenant_id you can filter by.
CREATE TABLE IF NOT EXISTS mcp_tasks (
  id         UUID        PRIMARY KEY DEFAULT gen_random_uuid(),
  tenant_id  UUID        NOT NULL REFERENCES mcp_tenants(id) ON DELETE CASCADE,
  title      TEXT        NOT NULL,
  status     TEXT        NOT NULL DEFAULT 'open' CHECK (status IN ('open', 'done')),
  created_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
  updated_at TIMESTAMPTZ NOT NULL DEFAULT NOW()
);

CREATE INDEX IF NOT EXISTS idx_mcp_tokens_tenant ON mcp_tokens (tenant_id);
CREATE INDEX IF NOT EXISTS idx_mcp_tasks_tenant  ON mcp_tasks  (tenant_id, created_at DESC);

MCP Postgres Starter

The one thing to understand about MCP auth​

Pick your tier​

What's included​

MCP tools​

Tier 2 (tasks.yml)​

Tier 2 — OIDC (auth-oidc.yml)​

Tier 1 (tasks-solo.yml)​

Auth lifecycle endpoints (Tier 2, auth.yml)​

Setup​

1. Managed variables​

2. Create the tables​

3. Deploy the configs for your chosen tier​

Connecting an MCP client​

Quick start (Tier 2 walkthrough)​

Auth, in depth​

Tier 1 — "for me": one token, full access​

Tier 2 — "for my customers": per-user tokens, tenant-scoped​

Using a real IdP (Auth0/Clerk/Cognito) — two options​

Revocation​

Customisation​

Notes & limitations​

Configuration​

auth-oidc.yml​

auth.yml​

seed.yml​

tasks-solo.yml​

tasks.yml​

schema.sql​