Spaces:
Running
Running
| from __future__ import annotations | |
| import contextvars | |
| from dataclasses import dataclass | |
| from datetime import datetime | |
| class AuthContext: | |
| api_key_id: str | |
| key_name: str | |
| key_prefix: str | |
| environment: str | |
| role: str | None | |
| scopes: frozenset[str] | |
| requests_per_minute: int | |
| concurrent_jobs: int | |
| uploads_per_hour: int | |
| processing_bytes_per_day: int | |
| expires_at: datetime | None | |
| # Resolved exclusively by the server from APIKeyPrincipal. These are not | |
| # accepted from REST, MCP, SDK, or n8n request payloads. | |
| workspace_id: str | None = None | |
| user_id: str | None = None | |
| membership_id: str | None = None | |
| membership_role: str | None = None | |
| def allows(self, required_scope: str) -> bool: | |
| if required_scope in self.scopes or "admin" in self.scopes: | |
| return True | |
| return False | |
| auth_context: contextvars.ContextVar[AuthContext | None] = contextvars.ContextVar( | |
| "auth_context", default=None | |
| ) | |
| # HTTP authentication already applies request and concurrency limits before the | |
| # MCP ASGI application runs. Standalone stdio MCP calls use this marker to apply | |
| # the same limiter exactly once inside the transport-neutral registry. | |
| http_auth_applied: contextvars.ContextVar[bool] = contextvars.ContextVar( | |
| "http_auth_applied", default=False | |
| ) | |