1"""The `MCPResource` base class. Subclass, set `uri` (or `uri_template`), and implement `read()`."""
  2
  3from __future__ import annotations
  4
  5import re
  6from abc import ABC, abstractmethod
  7from typing import TYPE_CHECKING, Any, get_type_hints
  8
  9if TYPE_CHECKING:
 10    from .views import MCPView
 11
 12
 13class MCPResource(ABC):
 14    """Base class for MCP resources.
 15
 16    Resources are addressable data sources identified by a URI that
 17    clients can list and read. Metadata comes from class attributes;
 18    content from `read()`:
 19
 20        class AppVersion(MCPResource):
 21            '''Current deployed version.'''
 22
 23            uri = "config://app/version"
 24            mime_type = "text/plain"
 25
 26            def read(self) -> str:
 27                return settings.VERSION
 28
 29    `read()` may return `str` (emitted as `text`) or `bytes` (emitted as
 30    base64 `blob`). Resource instances have `self.mcp` set by the
 31    dispatcher before `read()` is called — use it to read the caller's
 32    user, request, etc.
 33
 34    Override `allowed_for(mcp)` (classmethod) to filter when the resource
 35    is included — same pattern as `MCPTool`.
 36
 37    **URI templates.** For parametrized resources (one class, many URIs),
 38    set `uri_template` instead of `uri` and accept the template params on
 39    `__init__`:
 40
 41        class Order(MCPResource):
 42            '''An order by ID.'''
 43
 44            uri_template = "orders://{order_id}"
 45            mime_type = "application/json"
 46
 47            def __init__(self, order_id: int):
 48                self.order_id = order_id
 49
 50            def read(self) -> str:
 51                return str(Order.query.get(pk=self.order_id))
 52
 53    Templates match RFC 6570 level 1 — `{name}` placeholders match a
 54    single path segment (no slashes). Extracted params are coerced to the
 55    `__init__` annotation when it's `int`, `float`, or `bool`; otherwise
 56    passed through as strings.
 57    """
 58
 59    uri: str = ""
 60    uri_template: str = ""
 61    name: str = ""
 62    description: str = ""
 63    mime_type: str = ""
 64
 65    _uri_pattern: re.Pattern[str] | None = None
 66    _init_hints: dict[str, Any] | None = None
 67
 68    # Set by the MCPView dispatcher before `read()` is called.
 69    mcp: MCPView
 70
 71    def __init__(self) -> None:
 72        """Default no-arg init — template resources override with typed params."""
 73
 74    @abstractmethod
 75    def read(self) -> str | bytes:
 76        """Return the resource contents (str → text, bytes → base64 blob)."""
 77
 78    @classmethod
 79    def allowed_for(cls, mcp: MCPView) -> bool:
 80        """Return False to exclude this resource from `mcp`'s resource set.
 81
 82        Resources that return False are hidden from `resources/list`,
 83        `resources/templates/list`, and rejected from `resources/read` (as
 84        "unknown resource" — existence isn't leaked).
 85        """
 86        return True
 87
 88    @classmethod
 89    def matches(cls, uri: str) -> dict[str, Any] | None:
 90        """If this resource can serve `uri`, return params for `__init__`.
 91
 92        Returns `{}` for static URIs that match, `{param: value}` for
 93        template matches (values coerced via `__init__` annotations for
 94        `int`/`float`/`bool`; strings otherwise), or `None` if the URI
 95        doesn't match. Raises `ValueError` if the regex matches but
 96        coercion of the extracted params fails.
 97        """
 98        if cls.uri:
 99            return {} if uri == cls.uri else None
100        if cls._uri_pattern is not None:
101            match = cls._uri_pattern.fullmatch(uri)
102            if match is None:
103                return None
104            return _coerce_template_params(cls, match.groupdict())
105        return None
106
107    def __init_subclass__(cls, **kwargs: Any) -> None:
108        super().__init_subclass__(**kwargs)
109        if not cls.__dict__.get("name"):
110            cls.name = cls.__name__
111        if not cls.__dict__.get("description"):
112            doc = (cls.__doc__ or "").strip()
113            if doc:
114                cls.description = doc.splitlines()[0].strip()
115
116        if cls.uri and cls.uri_template:
117            raise TypeError(
118                f"{cls.__name__} must set only one of `uri` or `uri_template`"
119            )
120        if cls.uri_template:
121            cls._uri_pattern = _compile_uri_template(cls.uri_template)
122            try:
123                cls._init_hints = get_type_hints(cls.__init__)
124            except (NameError, TypeError):
125                # Unresolvable forward refs: skip coercion, pass raw strings.
126                cls._init_hints = {}
127
128
129_PLACEHOLDER = re.compile(r"\{([^{}]+)\}")
130
131
132def _compile_uri_template(template: str) -> re.Pattern[str]:
133    """Compile an RFC 6570 level-1 URI template into a regex.
134
135    Each `{name}` placeholder matches one path segment (no slashes).
136    """
137    pattern = ""
138    last = 0
139    for m in _PLACEHOLDER.finditer(template):
140        pattern += re.escape(template[last : m.start()])
141        pattern += f"(?P<{m.group(1)}>[^/]+)"
142        last = m.end()
143    pattern += re.escape(template[last:])
144    return re.compile(pattern)
145
146
147def _coerce_template_params(
148    cls: type[MCPResource], raw: dict[str, str]
149) -> dict[str, Any]:
150    hints = cls._init_hints or {}
151    coerced: dict[str, Any] = {}
152    for name, value in raw.items():
153        hint = hints.get(name)
154        if hint is int:
155            coerced[name] = int(value)
156        elif hint is float:
157            coerced[name] = float(value)
158        elif hint is bool:
159            coerced[name] = value.lower() in ("true", "1", "yes")
160        else:
161            coerced[name] = value
162    return coerced