koda-mcp-access

Reusable FastMCP access-control middleware for KODA-ecosystem MCP servers, plus identity helpers for tools. It is a pure Policy Enforcement Point (PEP): it reads identity + an already-resolved permission subset from proxy-injected headers and allows / denies / filters MCP tool calls by pattern matching.

It does not know about roles, servers, Okta, or the permissions model — that intelligence lives upstream (usrv-koda = data, koda-proxy = subset resolution). Any MCP behind a proxy that honors the header contract can use it unchanged.

How it fits

client → registry/nginx → koda-proxy → AgentCore runtime → your MCP + KodaAccessMiddleware
                          (resolves &                        (this package:
                           injects X-Koda-* )                 reads & enforces)

• Decision (who are you, what may you do) happens upstream in the koda-proxy: it calls usrv-koda, resolves the per-server subset, and injects it as base64-JSON in the permissions header.
• Enforcement (allow / deny / filter each call) happens here. This package never validates the JWT, never queries a DB, never knows roles.
• Trust: with signed identity on, the runtime only reads headers koda-proxy signed for it.

Components (C3)

What runs where, and which step each component owns:

 MCP client (Claude, Cursor, …)
   │  POST /mcp-koda/mcp   Authorization: Bearer <Okta JWT>
   ▼
┌─ mcp-gateway-registry ───────────────────────────────────────────────────────┐
│ nginx                                                                        │
│   ├─ auth_server /validate     JWT valid? groups                             │
│   └─ X-Koda-Server-Name = mcp-koda       (server path; the client cannot     │
│      X-Koda-Runtime-Arn = proxy_pass_url  set these, nginx overwrites them)  │
│                     │                                                        │
│                     ▼                                                        │
│ koda-proxy                                                                   │
│   1 verify the bearer (auth_server)                                          │
│   2 (server, ARN) is a registered pair ──────────► registry /api/servers     │
│   3 subset = permissions.mcp["mcp-koda"] ────────► usrv-koda profile API     │
│   4 build Koda-* headers, sign with k_mcp-koda ─► Secrets Manager            │
│                                                    .../mcp-identity          │
│   5 SigV4 InvokeAgentRuntime(ARN)                                            │
└─────────────────────┬────────────────────────────────────────────────────────┘
                      ▼
┌─ AgentCore runtime (one per MCP) ────────────────────────────────────────────┐
│ AgentCore forwards only the allowlisted ...-Custom-Koda-* headers            │
│                     │                                                        │
│                     ▼                                                        │
│ KodaAccessMiddleware (this package)                                          │
│   6 signature: audience, method, tool, ts ──► Secrets Manager                │
│                                               .../mcp-identity (same one),   │
│                                               derives k_mcp-koda             │
│   7 HeaderContract → Identity + MCPPermissions                               │
│   8 matcher: gate tools/call, filter tools/list                              │
│   9 access log line + audit_hook                                             │
│                     │                                                        │
│                     ▼                                                        │
│ your tools ── get_current_identity() / get_current_access_token()            │
└──────────────────────────────────────────────────────────────────────────────┘

Steps 1–5 decide, steps 6–9 enforce. Step 6 runs with signature_mode verify (the default once KODA_SERVER_NAME is set) or require; with off (no server name) the runtime trusts the headers as they arrive, like 1.2.0 did.

Example: Ana calls mcp-koda

Ana has role qa, which in usrv-koda grants:

permissions: { mcp: { "mcp-koda": { tools: ["greeting", "jira_*"] } }, skills: [...], agents: {...} }

For tools/call greeting, koda-proxy sends this to the mcp-koda runtime (prefix X-Amzn-Bedrock-AgentCore-Runtime-Custom-Koda- omitted):

Sub:          ana@kushki
Roles:        qa
Permissions:  base64({"tools": ["greeting", "jira_*"], "skills": [...], "agents": {...}})
Client-Id:    0oa1koda
Signed-At:    1760400000
Key-Id:       2026-10
Signature:    v1=9f2c…   # HMAC(k_mcp-koda, v1 · ts · mcp-koda · tools/call · greeting · headers)

What the middleware does with each request (runtime in require):

Wiring a runtime

from fastmcp import FastMCP
from koda_mcp_access import GuardConfig, KodaAccessMiddleware, get_current_identity

mcp = FastMCP("mcp-koda")
# Reads KODA_SERVER_NAME → verify, keyring from ai-projects/koda-platform/mcp-identity
mcp.add_middleware(KodaAccessMiddleware(GuardConfig()))


@mcp.tool()
async def greeting() -> str:
    return f"Hola, {get_current_identity().email}"

KODA_SERVER_NAME is the audience and the derive_for of the key: it must be the name koda-proxy sends as X-Koda-Server-Name, or every signature fails (reason=bad-signature). The secret is read by name, so each runtime gets the replica in its own region. At startup the middleware loads the keyring once and logs authz: signature_mode='verify' audience=mcp-koda-dev keyring=ai-projects/koda-platform/mcp-identity kid=…, or the error if the role cannot read it.

The signing secret is one for the whole platform, in each region where a reader runs (koda-proxy and every runtime; Secrets Manager replication keeps the copies equal):

{"kid": "2026-10", "key": "<32 random bytes, base64>"}

Rollout: set KODA_SERVER_NAME → verify (an invalid signature is denied, a missing one only logs WARNING authz: signature missing on <method>) → 24–48 h with no signature missing lines → KODA_SIGNATURE_MODE=require. KODA_SIGNATURE_MODE=off turns it off.

Install

uv add koda-mcp-access
# or with the optional standalone Okta verifier (local-dev):
uv add "koda-mcp-access[okta]"
# or: pip install koda-mcp-access
# release candidate (pip skips pre-releases unless pinned):
uv add "koda-mcp-access==1.3.0rc1"

Pin it in your project as:

# pyproject.toml
koda-mcp-access>=1.3,<2

Use

from fastmcp import FastMCP
from koda_mcp_access import KodaAccessMiddleware

mcp = FastMCP("My MCP")
mcp.add_middleware(KodaAccessMiddleware())   # defaults: AgentCore prefix, fail-closed

With an audit hook for observability (every allow/deny decision):

from koda_mcp_access import GuardConfig, KodaAccessMiddleware

def audit(d):
    logger.info("access action=%s target=%s decision=%s roles=%s",
                d.action, d.target, d.decision, d.identity.roles if d.identity else [])

mcp.add_middleware(KodaAccessMiddleware(GuardConfig(audit_hook=audit)))

Only tools are enforced. The middleware filters tools/list and gates tools/call against the caller's tools grant. Skills are authorized upstream in the registry, not here. Prompts are deprecated and not part of the permission model. Resources pass through (see Behavior).

Read identity from inside a tool:

from koda_mcp_access import get_current_identity

@mcp.tool()
async def whoami() -> dict:
    ident = get_current_identity()
    return {"sub": ident.sub, "roles": ident.roles, "tenant": ident.tenant_id}

Tools that call a backend on the caller's behalf can get the raw JWT as a FastMCP AccessToken (or a compact user dict):

from koda_mcp_access import get_current_access_token, get_current_user

@mcp.tool()
async def list_my_initiatives() -> list:
    token = get_current_access_token()        # None if unauthenticated
    return await backend.get("/initiatives", bearer=token.token)

Standard: every KODA-ecosystem MCP authorizes with a single add_middleware(KodaAccessMiddleware()) and no local src/auth/. If a tool needs identity, import it from this package. See docs/MCP_ACCESS_STANDARD.md.

Standalone local-dev (validate the Okta JWT itself, requires [okta] extra):

from koda_mcp_access.verifiers.okta import OktaTokenVerifier
from fastmcp.server.auth import RemoteAuthProvider

if IS_LOCAL:
    mcp.auth = RemoteAuthProvider(
        token_verifier=OktaTokenVerifier(),
        authorization_servers=[OKTA_ISSUER],
        base_url=MCP_BASE_URL,
    )

Header contract

The upstream proxy injects, under a configurable prefix (default x-amzn-bedrock-agentcore-runtime-custom-koda-):

permissions decodes to a flat object — the server-scoped tools plus the global catalogs skills / agents:

{ "tools": ["my_profile"],
  "skills": ["pm_challenge"], "agents": { "roadmap-agent": { "actions": ["*"] } } }

Behavior

Only tools are role-gated. Resources pass through for any authenticated caller (the permission model has no resource dimension); the fail-closed check still applies (no identity → denied). Skills are authorized upstream in the registry. Prompts are deprecated and not part of the permission model: they have no hook of their own and fall under the generic on_request check.

Discovery only lists. The discovery header lets the registry scan the catalogue. It never lets a request execute a tool or read content: on tools/call, resources/read and every other request it is ignored.

Access log

Every operation logs one line on the koda_mcp_access.middleware logger: who came in and what they used. Allows go to INFO; denials and fail-open calls go to WARNING.

INFO    access sub=ana@kushki roles=qa tenant=acme client=0oa1koda action=tools/call target=greeting decision=allow
WARNING access sub=ana@kushki roles=qa tenant=acme client=0oa1koda action=tools/call target=my_profile decision=deny
WARNING access sub=- roles=- tenant=- client=- action=tools/call target=greeting decision=deny reason=missing-identity
WARNING access sub=- roles=- tenant=- client=- action=tools/call target=greeting decision=deny reason=bad-signature
INFO    access sub=- roles=- tenant=- client=- action=tools/list target=* decision=allow bypass=discovery

• action is the MCP method; target is the tool name, resource URI or prompt name (* for lists).
• bypass (discovery, transport, fail_open) and reason (missing-identity, missing-permissions, or a signature reason: missing-signature, bad-signature, stale-signature, bad-timestamp, unknown-key, keyring-unavailable) appear only when they apply.
• Signature denials log sub=-: unsigned headers are not trusted, not even for the log.
• It never logs the bearer, the grants or the tool arguments. Whitespace and control characters in values are replaced with _ so a caller cannot forge log lines.
• audit_hook receives the same decisions (with action = MCP method) if you need to ship them somewhere else. Turn the line off with GuardConfig(access_log=False).

Signed identity context

Without it, anyone allowed to call the runtime directly (InvokeAgentRuntime) can send made-up Koda-* headers. With it, koda-proxy signs them and the runtime only trusts what it signed. The signature proves who sent the headers and that nobody changed them; it does not encrypt them.

# env: KODA_SERVER_NAME=mcp-koda  →  verify with the shared secret
mcp.add_middleware(KodaAccessMiddleware(GuardConfig()))

# the same, explicit (arguments win over the env):
from koda_mcp_access import SecretsManagerKeyring

mcp.add_middleware(KodaAccessMiddleware(GuardConfig(
    signature_mode="require",  # off | verify | require
    signature_audience="mcp-koda",  # this runtime's server name
    signing_key_provider=SecretsManagerKeyring(
        "ai-projects/koda-platform/mcp-identity", derive_for="mcp-koda"
    ),
)))

• Checked on every request but initialize / ping (and stdio), discovery included, before any Koda-* header is read.
• Each runtime verifies with its own key, derived from the shared secret (derive_for), so a signature made for one runtime does not verify on another. The signature also covers the audience (server name), the MCP method, the tool name on tools/call and a timestamp (max_skew_s, default 300).
• Whoever reads the secret can sign for any runtime. Grant GetSecretValue only to koda-proxy and the runtimes' roles.
• SecretsManagerKeyring (the default provider) caches the secret 5 minutes and accepts the current and the previous kid, so keys rotate without a redeploy. Rotate in two steps: new key in previous first; after 5 minutes, swap it into key (old one in previous), so no runtime sees a kid it does not know yet. The secret is {"kid", "key", "previous": {"kid", "key"}}, key in base64.
• require cannot be combined with fail_mode="open" (ValueError at startup).
• AgentCore only forwards allowlisted headers: add ...-Custom-Koda-Signed-At, ...-Key-Id, ...-Signature and ...-Client-Id to the runtime, or require denies everything.

The contract (canonical message, key derivation, test vector) is in docs/MCP_ACCESS_STANDARD.md.

Pattern matching

Every grant list is matched the same way (: and . are interchangeable separators). An empty / missing list always denies (fail-closed):

agents is a map {name: {actions: [...]}}; is_agent_action_allowed matches the agent key (exact / * / prefix), then the action against that entry's list.

Configuration (GuardConfig)

Auto-enrollment (opt-in)

When auto_enroll_group is set, a caller with no role (empty roles) is added to that Okta group. This is the one path where the package calls the Okta Management API — it requires the [okta] extra and admin credentials via env:

mcp.add_middleware(KodaAccessMiddleware(GuardConfig(
    auto_enroll_group="koda-builder",
)))

The service app must use private_key_jwt auth and hold the okta.groups.manage scope. A web-app client_id + client_secret cannot manage groups. OKTA_MGMT_* overrides OKTA_* when both are set.

The write (PUT /api/v1/groups/{gid}/users/{uid}) is idempotent; enrollment is best-effort (failures are logged, never raised) and eventual — the new membership takes effect on the caller's next session, once the upstream authorizer/proxy re-resolves roles. Leaving auto_enroll_group=None (the default) keeps the package a pure PEP with zero Okta writes.

Note: only callers with no role at all (empty roles header) are enrolled; anyone carrying any role is left untouched.

Internals

One file per concern — all role/server-agnostic:

Request flow inside on_call_tool: bypass if stdio → verify the signature (unless off) → resolve identity + perms from headers (the discovery flag is ignored here) → deny if tool not granted → call_next → reset identity (always).

KODA-30 / AR-01 status

As of 2026-10-14. The live checklist is in the KODA-30 plan (planes/koda-mcp-access/PLAN_KODA-30_DISCOVERY_BYPASS.md).

Pending:

• Publish 1.3.0 and move the consumers to it: usrv-mcp-koda, usrv-koda-infra-mcp, usrv-mcp-koda-builder (and its sqs_consumer), qmetry-mcp.
• The shared signing secret (ai-projects/koda-platform/mcp-identity) in each region.
• Each runtime: allowlisted headers, GetSecretValue on that secret, then verify → require.
• Least privilege on bedrock-agentcore:InvokeAgentRuntime (Phase 3).
• Sign the agent path too (koda-agent-runtime trusts Koda-* headers the same way).
• H3 (auto-enrollment on unverified identity), H4 (tags and Trusted Publishing), H5–H7 (low).

Develop

uv sync --group dev
uv run pytest
uv run ruff check src/ tests/