Coverage for src/ecnl/adapters/inbound/tools/_base.py: 87%
84 statements
« prev ^ index » next coverage.py v7.14.2, created at 2026-06-28 08:19 +0000
1"""Shared infrastructure for inbound MCP tool modules.
3`_safe_call` translates domain exceptions raised by the application service
4into readable error strings the LLM can present. `_READ_ANNOTATIONS` flags
5tools as read-only / idempotent so MCP clients can reason about them.
6`_authorize_tool` consults the wired :class:`ecnl.ports.inbound.Authorizer`
7before any tool dispatch — wired by the composition root in
8:mod:`ecnl.server` per the architecture roadmap's authorization port.
9"""
11import logging
12from collections.abc import Awaitable, Callable
14from mcp.types import ToolAnnotations
16from ....domain.exceptions import ECNLNotFoundError, UpstreamAPIError
17from ....ports.inbound import AuthorizationRequest, Authorizer
19logger = logging.getLogger(__name__)
21_READ_ANNOTATIONS_BASE = ToolAnnotations(
22 readOnlyHint=True,
23 destructiveHint=False,
24 idempotentHint=True,
25 openWorldHint=True,
26)
27"""Standard MCP hints. Kept separately for tools that need only the
28spec-defined fields without the architecture-roadmap extensions."""
31# ---------------------------------------------------------------------------
32# Extended annotations (architecture/docs/agentic-posture.md — sensitivity,
33# cost_class, rate_limit_class)
34# ---------------------------------------------------------------------------
37# Sensitivity classifications surfaced to MCP clients (LLM gateways,
38# audit aggregators, policy engines). The values are intentionally
39# coarse — adding a class is cheap; partitioning an existing one is a
40# behaviour change for every consumer that branched on it.
41SENSITIVITY_PUBLIC = "public"
42"""Data is freely available to anyone with access to the upstream API.
43Default for the ECNL tools that wrap AthleteOne's public endpoints."""
44SENSITIVITY_INTERNAL = "internal"
45"""Data is not personally identifiable but the deployment treats it
46as confidential (rate-limited per-tenant, behind a private gateway,
47etc.). Reserved for future tools that surface enriched analytics."""
48SENSITIVITY_PII = "pii"
49"""Tool returns personally identifiable information. Reserved — no
50ECNL tool exposes PII today; the class is registered up-front so a
51future addition does not require coordinating a rename."""
54# Cost classifications. A consumer can use these to throttle, batch, or
55# defer expensive tool calls without inspecting per-call latency.
56COST_FREE = "free"
57"""Tool is a no-op cache lookup; safe to invoke at any rate. Reserved
58for future static-content tools."""
59COST_METERED = "metered"
60"""Tool calls an upstream API that the operator pays for per request.
61Default for every ECNL tool — AthleteOne, season-discovery are all
62metered upstreams."""
63COST_BILLABLE = "billable"
64"""Tool triggers downstream charges that flow through Lago (ADR-0019).
65Reserved for tools that explicitly emit billing events; today every
66ECNL tool is read-only so this is unused but registered for the
67identity-platform side of the portfolio."""
70# Rate-limit classifications. These pair with the per-deployment rate
71# limiter — the actual ceilings live in operator config, not here.
72RATE_LIMIT_LOW = "low"
73"""Tool participates in a slow / cheap bucket — e.g. one call per
74second per agent. Reserved for the analytics endpoints once we
75introduce per-class buckets."""
76RATE_LIMIT_STANDARD = "standard"
77"""Tool participates in the default rate-limit bucket. Today every
78ECNL tool falls here."""
79RATE_LIMIT_PREMIUM = "premium"
80"""Tool participates in a high-throughput bucket reserved for clients
81that pay for elevated limits. Reserved."""
84def read_annotations(
85 *,
86 sensitivity: str = SENSITIVITY_PUBLIC,
87 cost_class: str = COST_METERED,
88 rate_limit_class: str = RATE_LIMIT_STANDARD,
89 title: str | None = None,
90) -> ToolAnnotations:
91 """Build a read-only ToolAnnotations extended with the
92 architecture-roadmap fields (``sensitivity``, ``cost_class``,
93 ``rate_limit_class``).
95 The MCP SDK's ToolAnnotations Pydantic model is configured with
96 ``extra="allow"`` so the extension fields land on the wire next to
97 the standard hints — clients that don't know about them ignore
98 them per RFC-style forward compatibility, while RAR-aware policy
99 engines can route on them.
101 Defaults match the dominant pattern for the ECNL tools (public
102 soccer data, metered upstream, standard rate bucket). Stricter
103 tools override per-field.
104 """
105 extras: dict[str, str] = {
106 "sensitivity": sensitivity,
107 "cost_class": cost_class,
108 "rate_limit_class": rate_limit_class,
109 }
110 return ToolAnnotations(
111 title=title,
112 readOnlyHint=True,
113 destructiveHint=False,
114 idempotentHint=True,
115 openWorldHint=True,
116 **extras,
117 )
120_READ_ANNOTATIONS = read_annotations()
121"""Default read-only annotation set surfaced to every ECNL tool.
123Every ECNL tool today wraps a read-only public soccer-data endpoint
124so the default sensitivity/cost/rate-limit classifications are
125correct. Tools with different characteristics override by calling
126:func:`read_annotations` directly with explicit values.
127"""
130async def _safe_call[T](coro: Awaitable[T], fmt: Callable[[T], str]) -> str:
131 """Await coro, apply fmt to the result, and convert domain exceptions to error strings.
133 Args:
134 coro: An awaitable returning the raw domain result.
135 fmt: A callable converting the domain result to a formatted string.
137 Returns:
138 The formatted string, or an error message if a domain exception was raised.
139 """
140 try:
141 return fmt(await coro)
142 except ECNLNotFoundError as exc:
143 logger.warning("Not found: %s", exc)
144 return f"Not found: {exc}"
145 except UpstreamAPIError as exc:
146 logger.error("Upstream API error: %s", exc)
147 return f"Upstream error: {exc}"
148 except ValueError as exc:
149 logger.warning("Invalid request: %s", exc)
150 return f"Invalid request: {exc}"
153async def _safe_call_authorized[T](
154 authorizer: Authorizer,
155 tool_name: str,
156 coro: Awaitable[T],
157 fmt: Callable[[T], str],
158) -> str:
159 """Authorize the call, then run :func:`_safe_call`.
161 Single dispatcher shared by every tool handler so the
162 "authorize → execute → format" pipeline stays consistent across the
163 20+ tools without each one repeating the wiring. A denial
164 short-circuits before the coroutine runs — no upstream fetch
165 happens for a forbidden call, which matters for both cost and
166 audit attribution.
167 """
168 denied = await _authorize_tool(authorizer, tool_name)
169 if denied is not None:
170 # Closing the coroutine releases anything it was holding (e.g.
171 # an async generator) so the upstream call never starts.
172 coro.close() if hasattr(coro, "close") else None
173 return denied
174 return await _safe_call(coro, fmt)
177async def _authorize_tool(authorizer: Authorizer, tool_name: str) -> str | None:
178 """Consult the authorizer for a tool dispatch.
180 Returns ``None`` when the call is allowed; the caller proceeds to
181 :func:`_safe_call`. Returns a "Forbidden: <reason>" string when
182 denied; the tool handler returns that string to the LLM so the
183 model can recover or surface the rejection to the user.
185 The caller's identity comes from the streamable-http transport's
186 AccessToken (set by the :class:`JWKSTokenVerifier` from
187 :mod:`ecnl.security`). The token's scopes carry the ADR-0015
188 ``actor_type``, ``agent_id``, and ``sub`` claims as
189 ``actor_type:<value>`` etc. — we parse them back into the
190 AuthorizationRequest here so the application layer doesn't see
191 the encoding.
193 The tool's architecture-roadmap annotations (sensitivity,
194 cost_class, rate_limit_class) are looked up from
195 :data:`_TOOL_ANNOTATIONS_REGISTRY` so the policy engine can branch
196 on the bucket without a separate catalog call. Missing entries
197 fall back to the public/metered/standard defaults.
198 """
199 actor = _read_actor()
200 annotations = _TOOL_ANNOTATIONS_REGISTRY.get(tool_name, _DEFAULT_TOOL_ANNOTATIONS_TUPLE)
201 decision = await authorizer.authorize(
202 AuthorizationRequest(
203 actor_type=actor["actor_type"],
204 agent_id=actor["agent_id"],
205 subject=actor["sub"],
206 client_id=actor["client_id"],
207 tool_name=tool_name,
208 sensitivity=annotations[0],
209 cost_class=annotations[1],
210 rate_limit_class=annotations[2],
211 )
212 )
213 if decision.allowed:
214 return None
215 logger.info(
216 "authz deny: tool=%s reason=%s actor=%s", tool_name, decision.reason, actor.get("actor_type") or "anonymous"
217 )
218 return f"Forbidden: {decision.reason}"
221# Per-tool annotation map populated at registration time. Keys are the
222# tool name (the @mcp.tool function name); values are the
223# (sensitivity, cost_class, rate_limit_class) triple. Default entries
224# are seeded lazily on first access via [register_tool_annotations].
225_TOOL_ANNOTATIONS_REGISTRY: dict[str, tuple[str, str, str]] = {}
226_DEFAULT_TOOL_ANNOTATIONS_TUPLE: tuple[str, str, str] = (
227 SENSITIVITY_PUBLIC,
228 COST_METERED,
229 RATE_LIMIT_STANDARD,
230)
233def register_tool_annotations(
234 tool_name: str,
235 *,
236 sensitivity: str = SENSITIVITY_PUBLIC,
237 cost_class: str = COST_METERED,
238 rate_limit_class: str = RATE_LIMIT_STANDARD,
239) -> None:
240 """Record the architecture-roadmap annotations for a tool.
242 Called from each ``register_*_tools`` function alongside the
243 ``@mcp.tool(annotations=...)`` decorator so the wire-level
244 surface (visible to MCP clients) and the policy-port surface
245 (visible to the :func:`_authorize_tool` lookup) stay aligned.
246 Calling with the defaults is a no-op — the registry only stores
247 overrides, so unannotated tools fall through to the constant
248 defaults at authorize time.
249 """
250 if (sensitivity, cost_class, rate_limit_class) == _DEFAULT_TOOL_ANNOTATIONS_TUPLE:
251 # Skip the dict write so the registry only carries overrides;
252 # the lookup falls back to the defaults via .get's default.
253 return
254 _TOOL_ANNOTATIONS_REGISTRY[tool_name] = (sensitivity, cost_class, rate_limit_class)
257def _read_actor() -> dict[str, str]:
258 """Pull the bearer-token-derived actor identity from the current request context.
260 Reads the AccessToken set by the mcp SDK's AuthenticationMiddleware,
261 parses the ``actor_type:`` / ``agent_id:`` / ``sub:`` scope
262 encodings the :class:`JWKSTokenVerifier` produces, and returns
263 them as a plain dict. Missing or unset values become empty strings
264 so downstream code can treat them uniformly — the stdio transport
265 and the streamable-http transport without auth both end up here
266 with empty fields, which the PassThroughAuthorizer happily allows.
267 """
268 actor = {"actor_type": "", "agent_id": "", "sub": "", "client_id": ""}
269 token = _current_access_token()
270 if token is None:
271 return actor
272 actor["client_id"] = token.client_id or ""
273 _merge_scope_claims(actor, token.scopes or [])
274 return actor
277def _current_access_token():
278 """Return the AccessToken for the in-flight request, or None.
280 Wraps the mcp SDK import so a stdio-only build that omits the
281 auth subpackage degrades gracefully instead of failing the
282 import. The function is its own helper so the gocyclo-style
283 counter on :func:`_read_actor` stays within the project's
284 cyclomatic-complexity cap.
285 """
286 try:
287 from mcp.server.auth.middleware.auth_context import get_access_token
288 except ImportError:
289 return None
290 return get_access_token()
293def _merge_scope_claims(actor: dict[str, str], scopes: list[str]) -> None:
294 """Mutate ``actor`` in place with the ``key:value`` scope encodings.
296 Scopes without a colon are ignored (those are the conventional
297 OAuth scope values like ``read`` or ``write``). Scopes whose key
298 is not one of the four known actor fields are also ignored so
299 unrelated scope conventions cannot accidentally overwrite
300 identity fields.
301 """
302 for scope in scopes:
303 if ":" not in scope:
304 continue
305 key, _, value = scope.partition(":")
306 if key in actor:
307 actor[key] = value