Skip to content

Middleware

Cluster: Uncategorised | Type: component | MCP Tools: 3

Overview

Shared request middleware for G6 server surfaces (MCP, REST). Provides Bearer API-key authentication, Redis-backed rate limiting, credit deduction for PAYG billing, tier-gated tool access control, and a parallel local-mode stack (license-based auth, in-process rate limiter, local usage tracking) for desktop deployments.

When to use:

  • Enforcing subscription tier gates on MCP/REST tool calls
  • Validating API keys before routing to downstream handlers
  • Deducting PAYG credits per-tool with fail-closed behaviour on DB error

Example:

from mvp.middleware import authenticate, check_rate_limit, is_tool_allowed

ctx = authenticate(authorization_header)     # -> UserContext | None
if not ctx:
    return 401

rl = check_rate_limit(ctx.api_key_id, ctx.tier)  # -> RateLimitResult
if not rl.allowed:
    return 429

if not is_tool_allowed(ctx.tier, "run_pipeline"):
    return 403

Works well with: security_gateway, observability

Read-only MCP introspection: middleware_status, middleware_tier_policy, and middleware_tool_cost. These tools do not run the middleware chain or mutate auth, rate-limit, or credit state. Responses use completion_state (verified, qualified-draft, blocked-escalated), warning_card, evidence, and request_id pass-through.

Public API

ApiKeyIdentity

Transport-neutral identity resolved from a G6 API key.

Field Type Default
customer_id int required
user_email str required
tier str required
api_key_id int required
is_admin bool False
request_id str ''
trace_id str ''

ApiKeyVerificationUnavailable(RuntimeError)

Raised when the verifier cannot reach its backing identity service.

ApiKeyVerifier(Protocol)

Interface shared by REST, MCP, and web-adjacent auth call sites.

Methods:

verify(raw_key: str, request_id: str = '', trace_id: str = '', surface: str = 'shared') -> ApiKeyIdentity | None

Return an identity for a valid key, None for an invalid key.

PostgresApiKeyVerifier

Verify SaaS API keys against the billing tables in PostgreSQL.

Constructor:

Parameter Type Default
pool_getter Callable[[], object] _get_pool

Methods:

verify(raw_key: str, request_id: str = '', trace_id: str = '', surface: str = 'shared') -> ApiKeyIdentity | None

healthcheck(surface: str = 'shared') -> bool

Verify the backing PostgreSQL pool can execute a lightweight query.

UserContext

Authenticated user context attached to MCP requests.

Field Type Default
customer_id int required
user_email str required
tier str required
api_key_id int required
is_admin bool False
request_id str ''
trace_id str ''

LicenseBlob

Decoded license payload.

Field Type Default
customer_id int required
email str required
tier str required
tools list[str] required
hw_hash str required
issued_at str required
expires_at str required
min_version str required
signature bytes required
raw_payload bytes required

MiddlewareInput(BaseModel)

Field Type Default
op str required
parameters dict[str, Any] Field(default_factory=dict)

MiddlewareOutput(BaseModel)

Field Type Default
op str ''
result dict[str, Any] Field(default_factory=dict)
message str ''
completion_state Literal['verified', 'qualified-draft', 'blocked-escalated'] 'qualified-draft'
warning_card dict[str, Any] Field(default_factory=dict)
evidence dict[str, Any] Field(default_factory=dict)
request_id str ''
run_id str ''

MiddlewareBlock(AIBlock)

AIBlock wrapper for the G6 middleware chain (auth, rate-limit, tier-gate, usage).

Methods:

infer(input: MiddlewareInput) -> Result[MiddlewareOutput]

RateLimitResult

Field Type Default
allowed bool required
limit int required
remaining int required
retry_after_seconds float required

UsageResult

Field Type Default
allowed bool required
balance_cents int required
cost_cents int required
error str ''

Functions

get_default_api_key_verifier() -> ApiKeyVerifier

Return the process-wide default verifier.

set_default_api_key_verifier(verifier: ApiKeyVerifier | None) -> None

Set or reset the process-wide verifier. Intended for integration tests.

check_api_key_verifier_health(verifier: ApiKeyVerifier | None = None, surface: str = 'shared') -> bool

Return True when the configured API-key verifier dependency is reachable.

verify_api_key(raw_key: str, request_id: str = '', trace_id: str = '', verifier: ApiKeyVerifier | None = None, surface: str = 'shared') -> ApiKeyIdentity | None

Verify a raw API key with the configured verifier.

verify_authorization_header(authorization_header: str | None, request_id: str = '', trace_id: str = '', verifier: ApiKeyVerifier | None = None, surface: str = 'shared') -> ApiKeyIdentity | None

Verify a standard Authorization: Bearer ... header.

authenticate(authorization_header: str, request_id: str = '', trace_id: str = '') -> Optional[UserContext]

Validate a Bearer token and return UserContext or None.

parse_license(data: bytes) -> LicenseBlob

Parse a license blob from JSON bytes + trailing signature.

check_expiry(blob: LicenseBlob) -> tuple[bool, bool]

Check license expiry.

authenticate_local(license_path: Path | None = None, decrypt_fn: Any = None, public_key: Any = None, current_hw_components: dict[str, str] | None = None, allow_unsigned: bool | None = None) -> UserContext | None

Read and validate cached license. No network call.

start_renewal_thread(api_key: str, license_path: Path | None = None, saas_url: str | None = None, interval: int = _RENEWAL_INTERVAL_SECONDS) -> None

Start a background thread that renews the license every 7 days.

stop_renewal_thread() -> None

Stop the background renewal thread.

check_rate_limit(api_key_id: int, tier: str) -> RateLimitResult

In-process sliding window rate limit check.

init_local_usage(db_path: str = ':memory:', initial_balance: int = 0, customer_id: int = 0) -> None

Initialize the local usage database with optional starting balance.

check_credits(customer_id: int, tool_name: str) -> UsageResult

Check if the customer has enough local credits for this tool.

deduct_credits(customer_id: int, tool_name: str, cost_cents: int) -> bool

Deduct credits and log usage locally.

log_usage(customer_id: int, tool_name: str, cost_cents: int | None = None) -> None

Log a tool invocation locally (no credit deduction, for non-PAYG tiers).

check_rate_limit(api_key_id: int, tier: str) -> RateLimitResult

Atomic sliding window counter rate limit check via Lua script.

build_transport_security_request(api_key: str | None, component: str, operation: str, params: Mapping[str, Any], source_ip: str, content_summary: str, transport: str, pipeline_steps: Sequence[Mapping[str, Any]] | None = None) -> SecurityRequest

Build a SecurityRequest for a transport-specific gateway wrapper.

is_tool_allowed(tier: str, tool_name: str) -> bool

Check if a tool is accessible for the given subscription tier.

requires_admin(tool_name: str) -> bool

Return True when a tool requires operator/admin authorization.

tool_cost_cents(tool_name: str) -> int

Return the billable PAYG cost for a tool invocation.

check_credits(customer_id: int, tool_name: str) -> UsageResult

Check if the PAYG customer has enough credits for this tool.

deduct_credits(customer_id: int, tool_name: str, cost_cents: int) -> bool

Deduct credits after successful tool execution and log usage.

log_usage(customer_id: int, tool_name: str, amount_cents: int = 0, surface: str = 'unknown', input_size_bytes: int = 0) -> bool

Log a successful non-PAYG tool invocation without deducting credits.

estimate_input_size_bytes(*args: Any, **kwargs: Any) -> int

Best-effort serialized input size for metering metadata.

record_successful_tool_call(customer_id: int, tier: str, tool_name: str, surface: str, input_size_bytes: int = 0, checked_cost_cents: int | None = None) -> bool

Record a successful tool call, deducting only for PAYG customers.

MCP Tools

Operation Source
verified middleware_mcp
qualified-draft middleware_mcp
blocked-escalated middleware_mcp