koda-mcp-authz-middleware 0.1.0

Reusable FastMCP authorization middleware (PEP) for KODA-ecosystem MCP servers.

koda-mcp-authz-middleware

Reusable FastMCP authorization middleware for KODA-ecosystem MCP servers. It is a pure Policy Enforcement Point (PEP): it reads identity + an already-resolved permission subset from proxy-injected headers and allows / denies / filters every MCP operation by pattern matching.

It does not know about roles, servers, Okta, or the grants permission 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.

Install

uv add koda-mcp-authz-middleware
# or with the optional standalone Okta verifier (local-dev):
uv add "koda-mcp-authz-middleware[okta]"
# or: pip install koda-mcp-authz-middleware

Pin it in your project as:

# pyproject.toml
koda-mcp-authz-middleware>=0.1,<0.2

Use

from fastmcp import FastMCP
from koda_mcp_authz_middleware import KodaAuthzMiddleware

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

With configuration (observability + prompts enforcement):

from koda_mcp_authz_middleware import GuardConfig, KodaAuthzMiddleware

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

mcp.add_middleware(KodaAuthzMiddleware(GuardConfig(
    enforce_prompts=True,
    audit_hook=audit,
)))

Read identity from inside a tool:

from koda_mcp_authz_middleware 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}

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

from koda_mcp_authz_middleware.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 — already server-scoped (tools/components) plus the global catalogs (skills/agents):

{ "tools": ["my_profile"], "components": ["ui://koda/*"],
  "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	per resource_gate	serve / AuthorizationError
resources/list	per resource_gate	filtered list
any	None (stdio / discovery)	passthrough (no filter)
any	header missing/invalid + not stdio/discovery	AuthorizationError (fail-closed)

resource_gate: by_mcp_access (default — any MCP access grants resources), by_component_list (strict URI match against components), or open.

Develop

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