Robert Helewka
93639188d3
- Rename MCPToken to UserToken across models, views, and tests - Update URL names from mcp-token-* to token-* - Add Daedalus/Pallas integration design doc (v2) - Switch docker-compose to build local mnemosyne:local image via shared build config instead of pulling from git.helu.ca
81 lines
3.1 KiB
Python
81 lines
3.1 KiB
Python
"""Helpers for accessing the request-scoped MCP auth state from inside tools.
|
|
|
|
The authoritative values are set by :class:`mcp_server.auth.MCPAuthMiddleware`
|
|
on the FastMCP ``Context``:
|
|
|
|
* ``STATE_KEY_USER`` — the Django user the bearer resolved to (synthetic
|
|
service user for JWT callers, concrete ``mcp_tokens.user`` for opaque
|
|
UserToken callers, ``None`` for team JWTs which are not tied to any
|
|
per-user account).
|
|
* ``STATE_KEY_TOKEN`` — the ``UserToken`` row for opaque-token callers;
|
|
``None`` for JWT callers.
|
|
* ``STATE_KEY_CLAIMS`` — the JWT claims dict for JWT callers; ``None``
|
|
for opaque-token callers. Intentionally exposed for debugging /
|
|
metrics; tools should NOT branch on claim shape for authorization,
|
|
they should read :func:`get_mcp_resolved_libraries` instead.
|
|
* ``STATE_KEY_RESOLVED_LIBRARIES`` — the authorization-resolved Library
|
|
UID set for this request. ``None`` means the caller is unauthenticated
|
|
or the auth middleware was bypassed (shouldn't happen in practice);
|
|
``[]`` means the caller is authenticated but scoped to zero libraries
|
|
(fail-closed); a non-empty list enumerates the UIDs the caller may read.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from fastmcp.server.context import Context
|
|
|
|
from .auth import (
|
|
STATE_KEY_CLAIMS,
|
|
STATE_KEY_RESOLVED_LIBRARIES,
|
|
STATE_KEY_TOKEN,
|
|
STATE_KEY_USER,
|
|
)
|
|
|
|
|
|
# ``Context.get_state`` is a synchronous method in FastMCP — it returns the
|
|
# stored value (``Any``) or ``None`` if the key is absent. Awaiting the
|
|
# returned value raises ``TypeError: object NoneType can't be used in 'await'
|
|
# expression`` whenever the value is ``None`` (and is semantically wrong even
|
|
# when it isn't). These helpers stay ``async def`` so call sites (and their
|
|
# ``await`` usage) don't have to change, but they call ``get_state``
|
|
# synchronously.
|
|
|
|
|
|
async def get_mcp_user(ctx: Context | None):
|
|
if ctx is None:
|
|
return None
|
|
return ctx.get_state(STATE_KEY_USER)
|
|
|
|
|
|
async def get_mcp_token(ctx: Context | None):
|
|
if ctx is None:
|
|
return None
|
|
return ctx.get_state(STATE_KEY_TOKEN)
|
|
|
|
|
|
async def get_mcp_claims(ctx: Context | None) -> dict | None:
|
|
"""Return the JWT claims dict for this request, or None for opaque-token callers."""
|
|
if ctx is None:
|
|
return None
|
|
return ctx.get_state(STATE_KEY_CLAIMS)
|
|
|
|
|
|
async def get_mcp_resolved_libraries(ctx: Context | None) -> list[str] | None:
|
|
"""Return the authorization-resolved Library UID list for this request.
|
|
|
|
Semantics (matching ``SearchRequest.resolved_libraries``):
|
|
|
|
* ``None`` — no auth information available (e.g. the middleware did
|
|
not run, or the tool was invoked outside a request context). Tools
|
|
should treat this as fail-closed and refuse to return content.
|
|
* ``[]`` — the caller was authenticated but is scoped to zero
|
|
libraries. Tools MAY proceed and return empty results.
|
|
* ``["lib_x", …]`` — the caller may read exactly these libraries.
|
|
|
|
See ``docs/DAEDALUS_PALLAS_INTEGRATION_v1.md`` §3.3 for the unified
|
|
auth model that populates this value.
|
|
"""
|
|
if ctx is None:
|
|
return None
|
|
return ctx.get_state(STATE_KEY_RESOLVED_LIBRARIES)
|