Skip to content

src/genro_asgi_multiworker_spa/spa_console.py

Source from this local checkout, regenerated when the reader rebuilds. Line links use #L<number>; a GitHub line range opens its first line.

1 # Copyright 2025 Softwell S.r.l.2 #3 # Licensed under the Apache License, Version 2.0 (the "License");4 # you may not use this file except in compliance with the License.5 # You may obtain a copy of the License at6 #7 #     https://www.apache.org/licenses/LICENSE-2.08 #9 # Unless required by applicable law or agreed to in writing, software10 # distributed under the License is distributed on an "AS IS" BASIS,11 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.12 # See the License for the specific language governing permissions and13 # limitations under the License.14 15 """The pool's debug door, served as MCP tools: ask a live server anything.16 17 One mechanism, no predicted questions: ``eval`` evaluates a Python18 expression in one process of an SPA pool — ``commander`` in the server19 process, or any worker by name, reached over the lane the commander already20 holds to every child — and answers the value's ``repr``. Whatever was not21 foreseen is readable by composing an expression; what a target's namespace22 holds is on :meth:`SpaConsole.eval`.23 24 **Mounting IS the gate.** The door is full eval by construction — there is no25 read-only eval in Python — so it exists only where the recipe mounts26 :class:`SpaConsoleMcpApplication` on purpose, and must never be mounted in27 production. An MCP client (Claude included) connects to the app's endpoint28 and asks in natural language; ``targets`` lists what can be looked into.29 """30 31 from __future__ import annotations32 33 from typing import Any34 35 from genro_routes import RoutingClass, route36 37 from genro_asgi.applications.mcp import McpApplication38 from .spa_app import SpaApplication39 40 __all__ = ["SpaConsole", "SpaConsoleMcpApplication"]41 42 43 class SpaConsole(RoutingClass):44     """The tool surface: every route is an MCP tool.45 46     Args:47         application: the MCP app this surface belongs to — its server is where48             the SPA fronts are found, at call time and never before (the tools49             run on a mounted, started server; the surface is built earlier).50     """51 52     def __init__(self, application: Any) -> None:53         self.route.plug("pydantic")54         self.application = application55 56     @route()57     async def targets(self) -> dict:58         """Every process the door can look into, by SPA application code."""59         return {60             code: front.commander.console_targets for code, front in self.spa_fronts.items()61         }62 63     @route()64     async def eval(self, expr: str, target: str = "commander", app: str = "") -> dict:65         """Evaluate a Python expression in one process of the pool.66 67         Args:68             expr: the expression; the namespace holds ``commander`` on the69                 vertex, ``worker`` inside a child.70             target: ``commander`` (default), or a worker's name from ``targets``.71             app: the SPA application code — needed only when the server mounts72                 more than one SPA front.73 74         Returns:75             The target and the value's ``repr``.76         """77         front = self.spa_front(app)78         return {"target": target, "repr": await front.commander.eval_in_target(target, expr)}79 80     @property81     def spa_fronts(self) -> dict[str, SpaApplication]:82         """The SPA fronts mounted on this server, by code."""83         return {84             code: mounted85             for code, mounted in self.application.server.applications.items()86             if isinstance(mounted, SpaApplication)87         }88 89     def spa_front(self, app: str) -> SpaApplication:90         """The front ``app`` names — or the only one, when the server has one.91 92         Raises:93             ValueError: no SPA front here, or several and ``app`` named none.94         """95         fronts = self.spa_fronts96         if app:97             if app not in fronts:98                 raise ValueError(f"no SPA front {app!r} — have: {', '.join(fronts) or 'none'}")99             return fronts[app]100         if len(fronts) == 1:101             return next(iter(fronts.values()))102         raise ValueError(103             f"several SPA fronts ({', '.join(fronts) or 'none'}): name one with app="104         )105 106 107 class SpaConsoleMcpApplication(McpApplication):108     """The MCP app whose whole tool surface is the pool's debug door.109 110     Recipe-friendly: mount it and the door exists, leave it out and it does111     not — mounting is the gate, and a production recipe never mounts it.112     """113 114     mcp_name = "genro-spa-console"115 116     def __init__(self, **kwargs: Any) -> None:117         super().__init__(routing_class=SpaConsole(self), **kwargs)