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 | |
| def allows(self, required_scope: str) -> bool: | |
| return "admin" in self.scopes or required_scope in self.scopes | |
| 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 | |
| ) | |