koda-mcp-access 1.0.1

Reusable FastMCP access-control middleware (PEP) + identity helpers for KODA-ecosystem MCP servers.

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.

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

Pin it in your project as:

# pyproject.toml
koda-mcp-access>=1.0,<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, so this package does not gate prompts/* or skill names. 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-):

Header suffix	Meaning
sub	caller subject (required; absent → no identity)
email	caller email (defaults to sub)
roles	space-separated roles (resolved upstream)
auth-groups	space-separated Okta groups
tenant-id	tenant
authorization	raw Okta JWT (for tools calling backends)
permissions	base64(JSON) — the already-resolved subset for this MCP
discovery	"true" → registry catalogue scan, no filtering

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

Operation	perms	Result
tools/call	match in tools	execute / AuthorizationError
tools/list	tools	filtered list
resources/read, resources/list, resources/templates/list	—	passthrough (not role-gated), still fail-closed without identity
any	None (stdio / discovery)	passthrough (no filter)
any	header missing/invalid + not stdio/discovery	AuthorizationError (fail-closed)

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, and prompts are not gated — this package does not touch prompts/*.

Pattern matching

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

Pattern	Matches
*	anything
jira_*	prefix (jira_create, jira_get, …)
my_profile	exact

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)

Field	Default	Purpose
header_prefix	AgentCore custom prefix	header namespace to read
fail_mode	"closed"	closed denies when identity/perms absent; open passes through
bypass_transports	{"stdio"}	transports that skip filtering (local dev)
discovery_suffix	"discovery"	header suffix that marks a catalogue scan
audit_hook	None	Callable[[AccessDecision], None] for every decision

Internals

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

Module	Responsibility
middleware.py	the Middleware hooks (on_call_tool, on_list_tools, on_*_resource*); binds identity per request, enforces, resets
headers.py	HeaderContract — builds header names from the prefix, reads them, base64-decodes permissions
identity.py	Identity dataclass + identity_from_headers
permissions.py	MCPPermissions shape + parse_permissions (tolerant of malformed input; ignores legacy prompts)
matcher.py	the pure pattern matchers; is_tool_allowed drives enforcement, is_skill_allowed / is_agent_action_allowed are exported helpers for consumers
context.py	contextvar holding the current Identity for the request
config.py	GuardConfig, AccessDecision
verifiers/okta.py	optional standalone Okta JWT verifier (local dev only)

Request flow inside on_call_tool: resolve identity + perms from headers → bypass if stdio/discovery → deny if tool not granted → call_next → reset identity (always).

Develop

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