nextQ MCP server

Read-only, multi-tenant MCP server that lets Claude answer seller questions from the research the nextQ pipeline already saved. It is built on FastMCP 4 and asyncpg, deployed on FastMCP Cloud, and authenticated with WorkOS AuthKit.

It reads only the aiml schema (the aimla_* views) of each tenant database. It never writes, and it never touches dwh.

The server's MCP instructions tell Claude which tool fits which question and how to answer, including the exact reply "table doesn't have this info". This works in any MCP client, so no skills are required. The skills in nextq-plugin/ are optional output templates.

How a request flows

MCP client ──HTTPS──▶ FastMCP Cloud ──▶ FastMCP server (stateless HTTP)
                                          │ 1. WorkOS JWT verified (signature, expiry, audience)
                                          │ 2. middleware: audit log, per-user rate limit, response cap
                                          │ 3. tenant resolved from the token (tenants.py), never from arguments
                                          │ 4. tenant's asyncpg pool (created lazily, shared by all its users)
                                          ▼
                                 tenant Postgres: aiml.aimla_* views, as nextq_reader

Concurrency.

• Every tool is async. Each tenant has its own asyncpg pool (NEXTQ_DB_POOL_MAX_SIZE connections), and a connection is held only for the length of one query.

• Independent queries run concurrently with asyncio.gather. One account loads with three parallel queries, and the search corpus with five.

• When many users ask for the same account at once, a single load runs and every caller shares it (single-flight TTL cache).

• Each tenant has its own search index, built in the background at startup. It is refreshed stale-while-revalidate, so a rebuild never blocks a request. The CPU-heavy part runs in a worker thread, keeping the event loop free.

• HTTP is stateless, so any instance can serve any request.

Security: read-only is enforced in three independent layers.

1. sql_guard.py parses model-written SQL with sqlglot. It allows only one SELECT over the catalog datasets, with no schema prefixes, writes, system functions or recursive CTEs, and it enforces a LIMIT.

2. Every session has default_transaction_read_only=on and a statement_timeout. Ad-hoc SQL also runs in an explicit READ ONLY transaction that is always rolled back.

3. The nextq_reader role (migrations/003_nextq_reader_role.sql) can only SELECT the aiml views.

Further protections:

• Tenant isolation: the tenant comes only from the verified token. Pools, caches and search indexes are all keyed by tenant. A user who matches no access rule gets no data.

• Errors: unexpected errors are masked (mask_error_details). Expected failures return a safe, actionable message.

• Logs: audit lines record the user, tool, outcome and latency. Tool arguments and results are never logged.

• Fail closed: the default environment is production, which refuses to start without WorkOS auth configured.

Related MCP server: MCPipe Toolchain

Layout

server.py                 FastMCP Cloud entrypoint (server.py:mcp)
nextq_mcp/
  app.py                  server factory: instructions, lifespan, middleware, /health
  config.py               NEXTQ_* settings (pydantic-settings), fail-closed validation
  auth.py                 WorkOS AuthKit / dev auth, caller identity
  tenants.py              tenant registry + user -> tenant resolution
  db.py                   per-tenant asyncpg pools, read-only execution
  queries.py              the server's own SQL over the aiml views
  runtime.py              app state (lifespan) + per-request TenantRuntime, error mapping
  cache.py                async single-flight TTL cache
  scoring.py              opportunity / role scoring (pure)
  search.py               BM25 knowledge search + per-tenant index manager
  catalog.py, sql_guard.py  ask-data catalog and SQL guard
  tools/                  the 12 tools (accounts, intel, ask_data, knowledge)
  data/                   catalog.yaml, scoring.yaml
nextq-plugin/             Claude Code plugin: remote server + optional skills
migrations/               nextq_reader role
evals/                    answerable / unanswerable / refuse question set
tests/                    113 tests: guard, scoring, search, tenants, cache, tools, HTTP auth

Tenant registry (NEXTQ_TENANTS)

A single JSON secret holds every tenant's database and the rules that map users to tenants. tenants.example.json is a dummy example.

{
  "tenants": {
    "c112": {
      "name": "NextQ demo",
      "database": {"host": "...", "port": 5432, "name": "qpilot_v1_c112",
                   "user": "nextq_reader", "password": "...", "sslmode": "require"},
      "aiml_schema": "aiml",
      "views": {}
    }
  },
  "access": {
    "user_ids":      {"user_01H...": "c112"},
    "emails":        {"seller@customer.com": "c112"},
    "org_ids":       {"org_01H...": "c112"},
    "email_domains": {"customer.com": "c112"}
  }
}

Rules are checked most specific first: WorkOS user id (sub), then email, then WorkOS organization (org_id), then email domain.

• views optionally overrides the view names per tenant (keys as in tenants.DEFAULT_VIEWS).

• Schema and view names must be plain lowercase identifiers.

• WorkOS access tokens carry sub and org_id by default. The emails and email_domains rules only work if you add an email claim with a WorkOS JWT template. Prefer org_ids: one WorkOS organization per customer.

Adding a tenant means updating the secret and redeploying.

Local development

Requires uv.

uv sync
cp .env.example .env              # dev settings: no auth, one tenant
# create tenants.local.json (gitignored) shaped like tenants.example.json
uv run pytest                     # 113 tests, no database needed
uv run ruff check . && uv run ruff format --check .

uv run python -m nextq_mcp                 # stdio
uv run python -m nextq_mcp --http          # http://127.0.0.1:8000/mcp
uv run fastmcp run server.py:mcp --transport http --port 8000   # exactly as FastMCP Cloud runs it

To use the local server from Claude Code over stdio: claude mcp add nextq-local -- uv run --directory /path/to/nextq-mcp python -m nextq_mcp

To test tenant routing over HTTP locally, set NEXTQ_AUTH_MODE=static with NEXTQ_DEV_TOKENS (see .env.example) and connect with Authorization: Bearer <token>. Settings refuse static and none when NEXTQ_ENVIRONMENT=production.

Deploy to FastMCP Cloud

1. WorkOS (one-time)

In the WorkOS dashboard:

1. Applications → Configuration: enable Dynamic Client Registration, so MCP clients (Claude, Cursor...) register themselves.

2. Resource Indicators: add https://<server-name>.fastmcp.app/mcp. FastMCP validates the token's aud against this exact URL.

3. Create an organization per customer and invite its users. Put each org_... id in access.org_ids.

4. Note your AuthKit domain, e.g. https://your-app.authkit.app.

See WorkOS' guide: https://workos.com/docs/authkit/mcp/integrating/token-verification

2. Database (per tenant, one-time)

Run migrations/003_nextq_reader_role.sql as an admin, and use nextq_reader in the registry. The database must accept TLS connections from FastMCP Cloud. FastMCP Cloud does not document fixed egress IPs, so an IP allowlist may not work.

3. FastMCP Cloud

1. Create a server from this GitHub repo:

   • Entrypoint: server.py:mcp

   • Authentication: OFF. The server runs WorkOS auth itself. FastMCP Cloud's built-in auth would consume the Authorization header, and the server could not see who the user is.

2. Set the environment variables (store the secrets as secrets):

   Variable	Value
   NEXTQ_PUBLIC_BASE_URL	https://<server-name>.fastmcp.app
   NEXTQ_WORKOS_AUTHKIT_DOMAIN	https://your-app.authkit.app
   NEXTQ_TENANTS	the registry JSON (secret)

   NEXTQ_ENVIRONMENT=production and NEXTQ_AUTH_MODE=workos are the defaults.

3. Deploy, then check https://<server-name>.fastmcp.app/health.

Dependencies come from pyproject.toml / uv.lock, and .python-version pins Python 3.12. FastMCP Cloud redeploys on every push to main. Protect main and require the CI workflow (.github/workflows/ci.yml) to pass before merging.

4. Connect

• Claude Code: claude --plugin-dir ./nextq-plugin (set NEXTQ_MCP_URL if the server name is not nextq), or claude mcp add --transport http nextq https://<server-name>.fastmcp.app/mcp.

• claude.ai / Claude Desktop: add a custom connector with the same URL.

On first use, the client opens the WorkOS login.

Configuration reference

Operations notes

• Connection budget: NEXTQ_DB_POOL_MAX_SIZE × number of instances must stay under the role's CONNECTION LIMIT (20).

• Memory: each tenant's search index lives in memory. The c112 index (~37k passages) builds in about 15 s. For very large tenants, move search.py to Postgres full-text search or pgvector behind the same tool.

• Per-instance state: caches, indexes and rate limits are per instance. The rate limit is a per-instance guard, not a global quota.

• Scores: opportunity and role scores rank signal strength. They are not win probabilities. Tune them in data/scoring.yaml.

• More data: to expose more datasets, add them to data/catalog.yaml with descriptions and add a matching GRANT to the migration.