"""`ctx.senpi_mcp` — the MCP client capability (runtime v2.0.0, Block B5). The author's `scan(inputs, ctx)` is SYNC. `ctx.senpi_mcp.call_tool(name, args)` is therefore a SYNC surface: the author never sees `await`. Internally it bridges the OFFICIAL MCP Python SDK (`mcp`), which is async-only: from mcp import ClientSession from mcp.client.streamable_http import streamablehttp_client async with streamablehttp_client(url, headers=...) as (read, write, _): async with ClientSession(read, write) as session: await session.initialize() result = await session.call_tool(name, arguments=args) DECISION (execution-plan B5, AMENDED 2026-09-04): use the SDK the STANDARD way — ONE persistent `ClientSession` created once (and `initialize()`d once), reused across every call, with no pooling toggle and no fresh-connection-per-call. The async context managers must STAY OPEN across calls, so we drive them on a dedicated event loop running in a background thread; `call_tool` submits work onto that loop and blocks for the result (the sync↔async bridge). AMENDMENT (RCA 2026-09-04, "External scanners go permanently blind after one MCP transport failure"): B5 originally also locked "no reconnect". In production the MCP server's `POST /mcp` stream is severed at the 60 s Kong upstream read timeout; the SDK then closes its streams but the client kept the dead session for the life of the process, so every later `call_tool` raised `ClosedResourceError` in ~1 ms until a runtime restart — 16–25 scanner MIDs blind per day, and authored `scan()` code swallowing the error made every tick report `ok` with zero candidates. The session is therefore now REBUILT: a transport-level failure (closed/broken stream, connection reset, broken pipe, httpx transport error) discards the dead session and the NEXT call opens and initialises a fresh one. The failing call still fails — there is no retry inside a call — the point is that the failure is no longer permanent. The failure is logged once as `senpi_mcp_error` with `session_rebuilt: true`. AMENDMENT, second half (same RCA): B5 also locked "no per-call timeout". `call_tool` blocked on the loop's future with no bound, so one stalled `POST /mcp` hung a call until Kong severed the stream at 60 s, and a handful of such calls consumed the whole 300 s tick budget (`scaffold_tick_timeout`). EVERY call is now BOUNDED — session open included — by `call_timeout_seconds` (launch env `SENPI_MCP_CALL_TIMEOUT_SECONDS`, default `DEFAULT_CALL_TIMEOUT_SECONDS`, well under the 60 s ceiling so the scaffold, not the proxy, is what gives up). A call that outlives the bound raises `SenpiMcpTimeoutError` (a `TimeoutError`) and is treated exactly like a transport failure: the abandoned session is discarded — the server may still write to that stream — and the next call opens a fresh one. It is logged as `senpi_mcp_error` with `error_type: TimeoutError`, `timeout_seconds` and `elapsed_ms` so a stall is distinguishable from a severed stream in ClickStack. AMENDMENT, third part (dev 2026-09-07): the SDK's transport is an anyio task group, so a failure raised through it arrives as an `ExceptionGroup` around the real cause. The client classifies and logs the LEAF, never the group: a wrapped closed stream still discards the session (the group's own type names no transport class, so without unwrapping the dead session would have been kept — the original wedge in a new coat), `error_type` is the leaf's class (never `ExceptionGroup`), and a sole-leaf group is re-raised to `scan()` as that leaf. Several leaves stay a group, logged as the first leaf plus `leaf_count`, and rebuild when ANY leaf is a transport failure. Response normalization MATCHES producer.mjs `createSenpiToolClient`: 1. isError truthy -> RAISE with the error text (never a silent None). 2. structuredContent present -> return it. 3. else first text content -> json.loads it (fallback to the raw text). 4. else -> return content as-is. 5. PLUS (scaffold-only, beyond producer.mjs): market-data numeric cast — Hyperliquid returns candle o/h/l/c/v and price-map values as STRINGS, and authored scanners doing arithmetic on them die on every tick. For the market tools (see _NUMERIC_CASTERS) the known numeric fields are cast to numbers before the payload reaches scan(); originals are kept under a sibling `_raw` key. Candle rows / price entries that do not cast are dropped — logged (`senpi_mcp_cast_dropped`, with asset and per-interval / per-key detail) AND marked on the payload (`_cast_dropped`) so scan() can react — matching the TS producer path (normalizeCandleRow + invalidRows in src/scanners/providers/adapter.ts). Everything else passes through byte-identical. The `mcp` package is LAZY-IMPORTED on first connect (NOT at module top), so `import scaffold.senpi_mcp` works even where `mcp` is not installed. Missing creds are LOUD AT FIRST USE: `call_tool` raises, construction never does. SESSION LIFETIME: each session lives inside ONE long-lived task on the loop (`_session_lifetime`), which enters the transport + ClientSession context managers, publishes the initialised session, then parks until a close is requested and exits the same context managers from the same task. That same-task shape is required: the SDK's transport is an anyio task group, and anyio refuses to exit a cancel scope from a task other than the one that entered it — so a dead session cannot be closed by calling `aclose()` from the `call_tool` task. A rebuild only flips the session's close event; the old lifetime task unwinds by itself, and `call_tool` never blocks on that teardown, so a wedged transport cannot hang a tick. TEARDOWN (intentional, one-process-per-scanner model): there is NO explicit close()/shutdown for the client. The background loop runs on a DAEMON thread, so it dies with the process. The scaffold runs exactly ONE client per process for that process's whole life — the supervisor restarts by spawning a FRESH child process (not by re-running run_scaffold in-place), so a single client is never replaced in-process and there is no loop leak across restart. On graceful stop (SIGTERM/SIGINT) run_scaffold exits its loop and the process returns; on a hard kill the OS reclaims the thread. We deliberately do NOT drive a close from a shutdown hook: it buys nothing over the dropped TCP connection (streamable-http is request/response, no protocol goodbye). """ from __future__ import annotations import asyncio import concurrent.futures import json import math import threading import time from typing import Any, Iterable, Optional from .loop_primitives import log_event # -------------------------------------------------------------------------- # Producer MCP write boundary — read-only by default. # -------------------------------------------------------------------------- # # A supervised producer's scan(inputs, ctx) runs in-process with the strategy's # full-capability SENPI_API_KEY. Without a boundary it could call ANY MCP tool — # including money/state mutations — bypassing every runtime guard (risk-guard, # DSL, slots, dedup). MUTATION_TOOLS enumerates those money/state-mutating tools; # call_tool BLOCKS them by default unless the EXACT tool name is on the client's # write_allowlist (least privilege — one permitted tool does not unlock others). # # Read tools (discovery_*, leaderboard_*, market_*, strategy_get*, strategy_list, # execution_get*, account_get*, user_get*, arena_*, audit_*, get_*, # list_senpi_guides, read_senpi_guide, ratchet_stop_get/list/events) are NOT in # this set and therefore always pass. MUTATION_TOOLS: frozenset[str] = frozenset( { "create_position", "close_position", "edit_position", "cancel_order", "send_usdc", "transfer_spot_to_perps", "strategy_create", "strategy_create_custom_strategy", "strategy_close", "strategy_close_positions", "strategy_update", "strategy_pause", "strategy_top_up", "strategy_withdraw_funds", "strategy_bridge_funds_from_hyperliquid_to_evm", "ratchet_stop_add", "ratchet_stop_edit", "ratchet_stop_delete", "user_claim_referral_rewards", } ) # THE allowlist a scaffold maintainer edits to permit a mutation tool for # producers. EMPTY BY DEFAULT — every MUTATION_TOOLS entry is blocked. The ONLY # way to permit a mutation is to add its exact tool name to this frozenset # (e.g. {"create_position"}). This is scaffold-maintainer-owned source: it is # not operator/env-configurable and is not reachable by authored producer code. PRODUCER_WRITE_ALLOWLIST: frozenset[str] = frozenset() # -------------------------------------------------------------------------- # Market-data numeric cast — scanner code receives numbers, not strings. # -------------------------------------------------------------------------- # # Hyperliquid's API (and therefore the Senpi MCP market tools) returns candle # o/h/l/c/v and price-map values as STRINGS. Every authored scanner must # otherwise rediscover this and hand-roll a float() helper; four independent # scanners shipped arithmetic-DOA on it in one review window. Casting here — # the single chokepoint every scanner market read passes through — makes # authored code quality irrelevant to that failure mode. # # Semantics mirror the TS producer path (asNumber in src/utils/response.ts + # normalizeCandleRow/invalidRows in src/scanners/providers/adapter.ts): # finite numbers pass through, numeric strings cast, candle rows whose # REQUIRED fields don't cast are dropped and counted. Enrichment sections # (asset_context, order_book, funding_history) cast valid values in place and # leave uncastable ones untouched — no data loss on optional context. # Originals of every rewritten section are preserved under a sibling `_raw` # key so precision-sensitive code can still reach the exact API strings. # Candle fields that MUST cast for the row to survive (t/o/h/l/c/v — the six # the TS normalizeCandleRow requires) vs. informational ones cast when valid. _CANDLE_REQUIRED_FIELDS = ("t", "o", "h", "l", "c", "v") _CANDLE_OPTIONAL_FIELDS = ("T", "n") # Known numeric fields of the Hyperliquid assetCtx embedded as asset_context. _ASSET_CONTEXT_NUMERIC_FIELDS = ( "markPx", "midPx", "prevDayPx", "oraclePx", "funding", "openInterest", "dayNtlVlm", "dayBaseVlm", "premium", ) def _as_number(value: Any) -> Optional[float]: """asNumber (src/utils/response.ts) ported: finite numbers pass through unchanged, numeric strings cast to float, everything else -> None. Deliberate divergences from JS: `""` is rejected (JS coerces it to 0), and integers beyond float range are rejected (json.loads yields arbitrary- precision ints; math.isfinite would raise OverflowError on them).""" if isinstance(value, bool): return None if isinstance(value, (int, float)): try: return value if math.isfinite(value) else None except OverflowError: return None if isinstance(value, str): try: number = float(value) except ValueError: return None return number if math.isfinite(number) else None return None def _record_candle_drop(stats: dict, interval: str, count: int = 1) -> None: stats["dropped_candle_rows"] += count stats["intervals"][interval] = stats["intervals"].get(interval, 0) + count def _cast_candle_series(series: list, interval: str, stats: dict) -> list: """New candle list with o/h/l/c/v (+t) numeric; uncastable rows dropped.""" out = [] for row in series: if not isinstance(row, dict): _record_candle_drop(stats, interval) continue cast_row = dict(row) ok = True for key in _CANDLE_REQUIRED_FIELDS: number = _as_number(row.get(key)) if number is None: ok = False break cast_row[key] = number if not ok: _record_candle_drop(stats, interval) continue for key in _CANDLE_OPTIONAL_FIELDS: if key in row: number = _as_number(row[key]) if number is not None: cast_row[key] = number out.append(cast_row) return out def _cast_if_numeric(container: dict, keys: Iterable[str]) -> dict: """Copy of `container` with each named field cast when castable; uncastable values stay as-is (enrichment sections never lose data).""" out = dict(container) for key in keys: if key in container: number = _as_number(container[key]) if number is not None: out[key] = number return out def _cast_asset_context(context: Any) -> Any: if not isinstance(context, dict): return context out = _cast_if_numeric(context, _ASSET_CONTEXT_NUMERIC_FIELDS) impact = context.get("impactPxs") if isinstance(impact, list): out["impactPxs"] = [ n if (n := _as_number(px)) is not None else px for px in impact ] return out def _cast_order_book(book: Any) -> Any: if not isinstance(book, dict) or not isinstance(book.get("levels"), list): return book out = dict(book) out["levels"] = [ [_cast_if_numeric(lvl, ("px", "sz")) if isinstance(lvl, dict) else lvl for lvl in side] if isinstance(side, list) else side for side in book["levels"] ] return out def _cast_asset_data(container: dict, stats: dict) -> dict: """market_get_asset_data: candles (drop-invalid) + enrichment sections.""" cast: dict[str, Any] = {} candles = container.get("candles") if isinstance(candles, dict): cast_candles: dict[str, Any] = {} for interval, series in candles.items(): if not isinstance(series, list): # A null/garbage series must not reach scan() as a string/None # hole in the guarantee — omit the key and count it, matching # the TS path (adapter.ts invalidRows on a non-array series). _record_candle_drop(stats, interval) continue cast_candles[interval] = _cast_candle_series(series, interval, stats) cast["candles"] = cast_candles if "asset_context" in container: new_context = _cast_asset_context(container["asset_context"]) if new_context is not container["asset_context"]: cast["asset_context"] = new_context if "order_book" in container: new_book = _cast_order_book(container["order_book"]) if new_book is not container["order_book"]: cast["order_book"] = new_book history = container.get("funding_history") if isinstance(history, list): cast["funding_history"] = [ _cast_if_numeric(entry, ("fundingRate", "premium")) if isinstance(entry, dict) else entry for entry in history ] return cast def _cast_prices(container: dict, stats: dict) -> dict: """market_get_prices: every value in the prices map; uncastable keys drop.""" prices = container.get("prices") if not isinstance(prices, dict): return {} out = {} for key, value in prices.items(): number = _as_number(value) if number is None: stats["dropped_price_keys"] += 1 if len(stats["price_keys"]) < _MAX_REPORTED_PRICE_KEYS: stats["price_keys"].append(key) continue out[key] = number cast: dict[str, Any] = {"prices": out} # Keep the MCP's sibling `count` true to the surviving map after drops. if stats["dropped_price_keys"] and "count" in container: cast["count"] = len(out) return cast _NUMERIC_CASTERS = { "market_get_asset_data": _cast_asset_data, "market_get_prices": _cast_prices, } # Cap on dropped price-key names carried in the log event / payload marker — # diagnosability without an unbounded list on a pathological response. _MAX_REPORTED_PRICE_KEYS = 10 def _cast_market_numerics(name: str, payload: Any, args: Optional[dict] = None) -> Any: """Apply the tool's numeric cast to an unwrapped payload (envelope-aware). Rewritten sections land back on the payload with their originals under `_raw` (written once, never clobbering a pre-existing key). Drops are visible three ways: the `senpi_mcp_cast_dropped` log event (with the asset and per-interval / per-key detail), a `_cast_dropped` marker on the container so scan() can react (the TS path's CandleResult.error analogue), and the originals under `_raw`. Anything unrecognized passes through untouched. """ caster = _NUMERIC_CASTERS.get(name) if caster is None or not isinstance(payload, dict): return payload # createResponse envelope: sections live under `data` when a boolean # `success` sits beside it (matches unwrapEnvelope in response.ts); bare # payloads keep sections at the top level. Error envelopes have no data # dict -> no-op. container = payload if isinstance(payload.get("success"), bool) and isinstance(payload.get("data"), dict): container = payload["data"] stats: dict[str, Any] = { "dropped_candle_rows": 0, "dropped_price_keys": 0, "intervals": {}, "price_keys": [], } cast_sections = caster(container, stats) if cast_sections: raw = {key: container[key] for key in cast_sections if key in container} container.update(cast_sections) if "_raw" not in container: container["_raw"] = raw dropped = {key: value for key, value in stats.items() if value} if dropped: container["_cast_dropped"] = dict(dropped) asset = (args or {}).get("asset") log_event( "senpi_mcp_cast_dropped", tool=name, **({"asset": asset} if isinstance(asset, str) else {}), **dropped, ) return payload # Per-call detail kept for one tick. The counts and the total duration below are # whole however many calls a tick makes; only this per-call list is cut, because # a scanner that calls a tool in a loop would otherwise grow an unbounded list # inside a process that must not grow. MAX_TICK_CALL_RECORDS = 64 # Per-argument bounds for a recorded call. Arguments are author-supplied and # reach a watching operator's screen, so what is kept is scalars only, capped. MAX_TICK_CALL_ARG_KEYS = 16 MAX_TICK_CALL_ARG_CHARS = 1024 # A list of scalars is kept as well, because the assets a scanner priced arrive # in that shape and are the most useful argument on the event. Both bounds apply # to one list: at most this many elements, and no more rendered characters than a # single string argument is allowed. MAX_TICK_CALL_ARG_CHARS is shared rather than # doubled so one long list cannot cost more frame than one long string. MAX_TICK_CALL_ARG_LIST_ITEMS = 16 def _is_arg_scalar(value: Any) -> bool: """Whether one value is a scalar this event carries: str, bool, or a finite number. The int branch covers bool, which subclasses it, and comes before the float branch on purpose: an int cannot be infinite, and `math.isfinite` on one too large to convert to a float raises OverflowError. """ if isinstance(value, int): return True if isinstance(value, float): return math.isfinite(value) return isinstance(value, str) def _record_arg_list(value: list) -> Optional[list]: """One list argument, bounded — or None when the list is not carried at all. A single non-scalar element rejects the whole list. Filtering it out instead would let an author's unbounded dict decide how much of a nested payload gets rendered, which is the input this event refuses to take. Elements dropped by either bound are replaced by a trailing count, so a shortened list reads as shortened rather than as a shorter call. The marker is a plain string element because the consumer expands dictionary indexes and scalars, and a nested object would not survive it. """ if not all(_is_arg_scalar(item) for item in value): return None kept: list = [] chars = 0 for item in value[:MAX_TICK_CALL_ARG_LIST_ITEMS]: rendered = len(item) if isinstance(item, str) else len(str(item)) if chars + rendered > MAX_TICK_CALL_ARG_CHARS: break chars += rendered kept.append(item) if len(kept) < len(value): kept.append(f"+{len(value) - len(kept)} more") return kept def _record_args(args: Optional[dict]) -> dict: """The scalar arguments of one call, bounded — never the caller's own dict. A reference would let authored code mutate what the tally already recorded, and a deep copy would pay for a payload the wire is going to drop anyway. A kept list is a new list for the same reason. """ if not isinstance(args, dict): return {} kept: dict = {} for key in list(args.keys())[:MAX_TICK_CALL_ARG_KEYS]: if not isinstance(key, str): continue value = args[key] if isinstance(value, str): kept[key] = value[:MAX_TICK_CALL_ARG_CHARS] elif _is_arg_scalar(value): # The same gate the list path applies, and for a harder reason here: an # `inf` compares equal to itself, so a self-equality check lets it # through, and json.dumps then writes the bare `Infinity` token. That # token is not JSON, and the tick facts ride one POST body — a # JavaScript JSON.parse on it throws and every other fact on the tick # is lost with the one argument. kept[key] = value elif isinstance(value, list): # An empty list is kept. Absent means unmeasured on this event, and a # scanner that passed no elements did measure something: it asked for # nothing. Dropping it would report the argument as never supplied. rendered_list = _record_arg_list(value) if rendered_list is not None: kept[key] = rendered_list return kept class TickErrorTally: """Per-tick record of what the MCP boundary did: every call, and every failure. Two tallies, one object, because both are read at the same moment and reset at the same moment. FAILURES. Authored scan() code routinely catches the MCP exception and returns [], so the exception alone never reaches the scaffold — the tick would finish `ok` with zero candidates while the scanner is blind. The client records every failure here as it re-raises; the scaffold resets the tally at tick start and reads it at tick end to decide whether a candidate-less tick was really ok. CALLS. Every call is recorded, succeeded or failed, with the time it took. Failures alone say what broke but not what the tick actually did or where its time went — a tick with no failures and one 9-second call is a slow venue, and nothing else on the box can tell an operator that. Thread-safe because call_tool may be driven from author threads. """ def __init__(self) -> None: self._by_type: dict[str, int] = {} self._calls: list[dict] = [] self._call_count = 0 self._total_ms = 0.0 self._lock = threading.Lock() def record(self, error_type: str) -> None: with self._lock: self._by_type[error_type] = self._by_type.get(error_type, 0) + 1 def record_call( self, tool: str, elapsed_ms: float, args: Optional[dict] = None, error_type: Optional[str] = None, ) -> None: """Record one completed call. Called on the success path and the failure path alike.""" entry = {"tool": tool, "ms": elapsed_ms, "args": _record_args(args)} if error_type is not None: entry["error_type"] = error_type with self._lock: self._call_count += 1 self._total_ms += elapsed_ms if len(self._calls) < MAX_TICK_CALL_RECORDS: self._calls.append(entry) def reset(self) -> None: with self._lock: self._by_type = {} self._calls = [] self._call_count = 0 self._total_ms = 0.0 def snapshot(self) -> dict[str, Any]: """{count, dominant_error_type, by_type}; count 0 / None when clean. The dominant type is the most frequent one, ties broken by first seen — so one dominant `ClosedResourceError` reads as the wedge it is even when a stray other failure sits beside it. """ with self._lock: by_type = dict(self._by_type) count = sum(by_type.values()) dominant = max(by_type, key=by_type.__getitem__) if by_type else None return {"count": count, "dominant_error_type": dominant, "by_type": by_type} def call_snapshot(self) -> dict[str, Any]: """{count, total_ms, calls}. `count` and `total_ms` are whole; `calls` may be cut.""" with self._lock: return { "count": self._call_count, "total_ms": round(self._total_ms, 1), "calls": [dict(call) for call in self._calls], } # -------------------------------------------------------------------------- # Transport-failure classification — which errors mean "the session is dead". # -------------------------------------------------------------------------- # # Matched by class NAME along the MRO so the module keeps its lazy-import # contract (no `anyio`/`httpx` import at module top). anyio raises the first # three when the SDK's streams are closed or broken under a session; httpx's # `TransportError` is the base of every connect/read/write/protocol/timeout # failure on the POST itself. OSError covers ConnectionError and its subclasses # (reset, refused, aborted), BrokenPipeError and — deliberately — TimeoutError, # which is what the per-call bound raises (`SenpiMcpTimeoutError`): a call that # timed out at the socket or under the bound has left the session in an unknown # state (the server may still write to the abandoned stream) and the next call # must not inherit it. # Our own pre-connect raises (PermissionError for a blocked mutation tool, # RuntimeError for missing creds) happen before any session is held, so they # never reach this classification. _TRANSPORT_ERROR_NAMES: frozenset[str] = frozenset( { "ClosedResourceError", "BrokenResourceError", "EndOfStream", "TransportError", } ) # The SDK's `streamablehttp_client` runs an anyio task group, so a failure # raised through it (a closed stream, an httpx error on the POST) surfaces as an # `ExceptionGroup` wrapping the real cause — observed on dev 2026-09-07 as # `error_type: ExceptionGroup, error: "unhandled errors in a TaskGroup (1 # sub-exception)"` against a 503. Classification and the fleet fingerprint must # look at the LEAVES: the group's own MRO names no transport class, so a wrapped # `ClosedResourceError` on a live session would otherwise keep the dead session # (the pre-rebuild wedge). Matched by name so the `exceptiongroup` backport on # Python < 3.11 unwraps like the builtin. _EXCEPTION_GROUP_NAME = "BaseExceptionGroup" def _is_exception_group(exc: BaseException) -> bool: return any(klass.__name__ == _EXCEPTION_GROUP_NAME for klass in type(exc).__mro__) def _leaf_exceptions(exc: BaseException) -> list[BaseException]: """Flatten an exception-group tree to its leaves; a plain exception is its own leaf.""" if not _is_exception_group(exc): return [exc] leaves: list[BaseException] = [] for sub in exc.exceptions: # type: ignore[attr-defined] leaves.extend(_leaf_exceptions(sub)) return leaves or [exc] def _unwrap_sole_leaf(exc: BaseException) -> BaseException: """The one real cause behind a group, else `exc` itself. A group with exactly one leaf is the anyio wrapping and nothing more, so the caller gets the leaf. Several leaves are genuinely several failures and stay a group — the log line then names the first leaf and carries the count. """ leaves = _leaf_exceptions(exc) return leaves[0] if len(leaves) == 1 else exc def _is_transport_error(exc: BaseException) -> bool: """True when `exc` means the MCP session's transport is unusable. Looks through exception groups: a group is a transport failure when ANY of its leaves is. """ return any(_leaf_is_transport_error(leaf) for leaf in _leaf_exceptions(exc)) def _leaf_is_transport_error(exc: BaseException) -> bool: if isinstance(exc, (OSError, EOFError)): return True return any(klass.__name__ in _TRANSPORT_ERROR_NAMES for klass in type(exc).__mro__) # -------------------------------------------------------------------------- # Per-call bound — the scaffold gives up before the proxy does. # -------------------------------------------------------------------------- # # Kong severs a stalled `POST /mcp` at 60.000 s; a tick has 300 s. 30 s is # comfortably under the ceiling — a stall is visible as OUR timeout, not as a # severed stream — and still leaves ten sequential calls inside one tick. The # bound covers the whole call: opening the session (initialize) when there is # none, plus the tool call itself. DEFAULT_CALL_TIMEOUT_SECONDS: float = 30.0 # Launch-env knob for the bound, read by scaffold._build_senpi_mcp. CALL_TIMEOUT_ENV_VAR = "SENPI_MCP_CALL_TIMEOUT_SECONDS" def is_valid_call_timeout(seconds: Any) -> bool: """A usable bound: a positive, finite number (bools are not numbers here). Shared by the constructor (which rejects anything else) and the launch-env reader in scaffold.py (which falls back to the default instead). """ if isinstance(seconds, bool) or not isinstance(seconds, (int, float)): return False return math.isfinite(seconds) and seconds > 0 class SenpiMcpTimeoutError(TimeoutError): """A `call_tool` outlived the client's per-call bound. A `TimeoutError` so authored `scan()` code can `except TimeoutError`; a distinct class so the scaffold's own bound is tellable from a socket timeout. Classified as a transport failure: the session it happened on is discarded and the next call opens a fresh one. """ class _McpSession: """One live MCP session: the SDK session plus the handles that end it. `lifetime` is the loop task that entered the SDK's context managers and will exit them once `close_requested` is set. `lifetime.done()` while the session is still held means the transport died on its own. """ __slots__ = ("session", "lifetime", "close_requested") def __init__( self, session: Any, lifetime: "asyncio.Task[None]", close_requested: asyncio.Event ) -> None: self.session = session self.lifetime = lifetime self.close_requested = close_requested @property def alive(self) -> bool: return not self.lifetime.done() class SenpiMcpClient: """Sync MCP client over the official async SDK with one persistent session. Construction is cheap and never touches the network or imports `mcp`. The first `call_tool` lazily imports the SDK, spins up a private event loop on a background thread (once, for the client's life), opens the transport + ClientSession, and initializes it once. Every subsequent call reuses that same session — until a transport failure, after which the dead session is discarded and the next call opens a fresh one (B5 amendment, RCA 2026-09-04). `tick_errors` is the per-tick failure tally the scaffold loop resets and reads (see TickErrorTally); every failure call_tool re-raises is recorded there exactly once, whatever its kind (transport, timeout, isError, ...). """ def __init__( self, api_key: Optional[str] = None, mcp_url: Optional[str] = None, write_allowlist: Iterable[str] = (), call_timeout_seconds: Optional[float] = None, ) -> None: self._api_key = api_key self._mcp_url = mcp_url if call_timeout_seconds is None: call_timeout_seconds = DEFAULT_CALL_TIMEOUT_SECONDS if not is_valid_call_timeout(call_timeout_seconds): raise ValueError( "senpi_mcp: call_timeout_seconds must be a positive number, " f"got {call_timeout_seconds!r}" ) self._call_timeout = float(call_timeout_seconds) # Mutation tools the operator has explicitly permitted. EMPTY by default # → every MUTATION_TOOLS entry is blocked (read-only producer boundary). self._write_allowlist: frozenset[str] = frozenset(write_allowlist) self.tick_errors = TickErrorTally() # Persistent bridge state. The loop + thread are created lazily on the # first call_tool and live for the client's life; the session is # (re)opened lazily whenever there is no live one. self._loop: Optional[asyncio.AbstractEventLoop] = None self._thread: Optional[threading.Thread] = None self._current: Optional[_McpSession] = None self._connect_lock = threading.Lock() # -- public sync surface ---------------------------------------------- @property def call_timeout_seconds(self) -> float: """The bound every `call_tool` fails within (session open included).""" return self._call_timeout def call_tool(self, name: str, args: Optional[dict] = None) -> Any: """Invoke an MCP tool synchronously; return the unwrapped result. Raises (loud, never a silent None) when the tool is a money/state mutation not on the write allowlist, when creds are missing, or when the tool reports `isError`. EVERY failure is logged (a `senpi_mcp_error` scaffold event) before it is re-raised, so even when the author's scan() swallows the exception in its own try/except, the failure still leaves a visible record at this boundary instead of vanishing into an empty result. A transport-level failure (see `_is_transport_error`) additionally discards the session it happened on, so the NEXT call opens a fresh one instead of failing forever; its log event carries `session_rebuilt: true`. The whole call — opening a session when there is none, then the tool call — must finish within `call_timeout_seconds`, else it raises `SenpiMcpTimeoutError` (logged with `error_type: TimeoutError` and the elapsed ms) and the abandoned session is discarded like any other transport failure. """ used: Optional[_McpSession] = None started = time.monotonic() deadline = started + self._call_timeout try: # Read-only producer boundary, enforced PRE-CONNECT (no transport # opened for a blocked write). A mutation tool passes only if its exact # name is on the write allowlist (least privilege — one tool ≠ all). if name in MUTATION_TOOLS and name not in self._write_allowlist: raise PermissionError( f'senpi_mcp: tool "{name}" is a money/state-mutating MCP tool and ' "is blocked — producers are READ-ONLY by default. The ONLY way to " f'permit it is to add "{name}" to the PRODUCER_WRITE_ALLOWLIST ' "frozenset in src/scanners/external/scaffold/senpi_mcp.py " "(scaffold-maintainer-owned source; not operator/env-configurable; " "not reachable by authored producer code)." ) used = self._ensure_connected(name, deadline) result = self._run(used.session.call_tool(name, arguments=args), name, deadline) value = _cast_market_numerics(name, self._unwrap(name, result), args) # Recorded after the unwrap, so a call that returned an `isError` result is # counted on the failure path below and never twice. self.tick_errors.record_call(name, round((time.monotonic() - started) * 1000, 1), args) return value except Exception as exc: # Catch every failure — mutation block, missing creds, the lazy # `import mcp`, transport errors, timeouts, tool `isError` — tally # it, log it, then re-raise to the author's scan(). Scoped to # Exception so a tick timeout (_TickTimeout, a BaseException) still # unwinds the tick untouched. This is the ONE exit for every failed # call, so the tally sees each failure exactly once. rebuilt = used is not None and _is_transport_error(exc) and self._discard(used) # Classify and log the LEAF, never the anyio ExceptionGroup around # it (see _leaf_exceptions): the fingerprint has to tell a 503 from # a closed stream. One leaf is the failure; several are logged as # the first leaf plus `leaf_count`. leaves = _leaf_exceptions(exc) primary = leaves[0] timed_out = isinstance(primary, SenpiMcpTimeoutError) # The fleet fingerprint for the bound is the builtin category, not # the scaffold subclass: ClickStack queries `TimeoutError`. The tally # counts under the SAME name, so the dominant type a blind tick # reports to /errors matches the senpi_mcp_error lines beside it. error_type = "TimeoutError" if timed_out else type(primary).__name__ elapsed_ms = round((time.monotonic() - started) * 1000, 1) self.tick_errors.record(error_type) self.tick_errors.record_call(name, elapsed_ms, args, error_type) log_event( "senpi_mcp_error", tool=name, error_type=error_type, error=str(primary)[:500], elapsed_ms=elapsed_ms, **({"leaf_count": len(leaves)} if len(leaves) > 1 else {}), **({"timeout_seconds": self._call_timeout} if timed_out else {}), **({"session_rebuilt": True} if rebuilt else {}), ) if len(leaves) == 1 and primary is not exc: # The author's scan() sees the real cause (so `except OSError` # or `except TimeoutError` keeps working), not the wrapper. raise primary raise # -- persistent session lifecycle ------------------------------------- def _ensure_connected(self, name: str, deadline: float) -> _McpSession: """Return the live session, opening one when there is none. A session whose lifetime task has ended on its own (the transport died between calls) counts as absent and is replaced here, before the call. Opening counts against the call's `deadline`: a stalled initialize raises `SenpiMcpTimeoutError` like a stalled tool call would. """ current = self._current if current is not None and current.alive: return current with self._connect_lock: current = self._current if current is not None and current.alive: return current # Loud at first use: creds are required to reach the MCP server. if not self._api_key: raise RuntimeError( "senpi_mcp: SENPI_API_KEY is not set; cannot call MCP tools " "(creds are resolved at launch from SENPI_API_KEY/SENPI_MCP_URL)." ) if not self._mcp_url: raise RuntimeError( "senpi_mcp: SENPI_MCP_URL is not set; cannot call MCP tools " "(creds are resolved at launch from SENPI_API_KEY/SENPI_MCP_URL)." ) if self._loop is None: # Dedicated event loop on a background thread, created ONCE per # client. The async context managers (transport + session) must # stay OPEN across calls, so we cannot use asyncio.run() per # call — that would tear the session down between calls. A # rebuild reuses this loop; it never spawns another thread. self._loop = asyncio.new_event_loop() self._thread = threading.Thread( target=self._loop.run_forever, name="senpi-mcp-loop", daemon=True, ) self._thread.start() self._current = self._run(self._open_session(), name, deadline) return self._current def _discard(self, dead: _McpSession) -> bool: """Drop `dead` so the next call opens a fresh session; ask it to close. Returns False (and touches nothing) when `dead` is no longer the held session — a concurrent caller already replaced it, and its successor must not be thrown away for the predecessor's failure. The close is fire-and-forget: the lifetime task exits its own context managers, and the caller never waits on a possibly-wedged transport. """ with self._connect_lock: if self._current is not dead: return False self._current = None self._loop.call_soon_threadsafe(dead.close_requested.set) return True async def _open_session(self) -> _McpSession: """Start a session lifetime task and wait until its session is initialised. A failure while opening propagates to the caller; the lifetime task has already unwound whatever it had entered, so nothing is left half-built. A cancelled open (the caller's bound ran out while initialize stalled) cancels the lifetime task too, which unwinds the transport from inside the task that entered it — a stalled open never leaks a session. """ loop = asyncio.get_running_loop() ready: "asyncio.Future[Any]" = loop.create_future() close_requested = asyncio.Event() lifetime = loop.create_task(self._session_lifetime(ready, close_requested)) try: session = await ready except asyncio.CancelledError: lifetime.cancel() raise return _McpSession(session, lifetime, close_requested) async def _session_lifetime(self, ready: "asyncio.Future[Any]", close_requested: asyncio.Event) -> None: """Own one session from open to close, all within this single task. Lazy-import the SDK here so `import scaffold.senpi_mcp` succeeds in envs without `mcp` installed (tests inject a fake `mcp` into sys.modules). """ try: from mcp import ClientSession from mcp.client.streamable_http import streamablehttp_client headers = {"Authorization": f"Bearer {self._api_key}"} async with streamablehttp_client(self._mcp_url, headers=headers) as (read, write, _): async with ClientSession(read, write) as session: await session.initialize() ready.set_result(session) await close_requested.wait() except BaseException as exc: if not ready.done(): # Opening failed: surface it to the waiting call_tool. The # transport's anyio task group wraps the cause in an # ExceptionGroup; the waiter gets the sole leaf (an httpx 503, # a refused connection) so it can classify and re-raise the # real error. A non-Exception (task cancellation) is wrapped # so the waiter gets a plain error instead of blocking forever. cause = _unwrap_sole_leaf(exc) ready.set_exception( cause if isinstance(cause, Exception) else RuntimeError(f"senpi_mcp: session open aborted by {type(cause).__name__}") ) # Otherwise the transport died under a published session (the # next call_tool sees `alive` False and rebuilds) or its teardown # complained after a rebuild — neither has a caller left to tell. if not isinstance(exc, Exception): raise def _run(self, coro: Any, name: str, deadline: float) -> Any: """Drive a coroutine on the persistent loop, blocking until `deadline`. Past the deadline the loop-side task is cancelled and `SenpiMcpTimeoutError` is raised. If the task finished in the instant between the deadline and the cancel, its outcome is taken as-is — the call made it, by a hair — so a completed open never leaks its session. """ future = asyncio.run_coroutine_threadsafe(coro, self._loop) try: return future.result(timeout=max(0.0, deadline - time.monotonic())) except concurrent.futures.TimeoutError: if not future.cancel(): return future.result() raise SenpiMcpTimeoutError( f'senpi_mcp: tool "{name}" did not complete within ' f"{self._call_timeout:g}s ({CALL_TIMEOUT_ENV_VAR}); the session (or " "the pending open) is discarded and the next call opens a fresh one" ) from None # -- response normalization (matches producer.mjs createSenpiToolClient) -- @staticmethod def _unwrap(name: str, result: Any) -> Any: if getattr(result, "isError", False): content = getattr(result, "content", None) if ( isinstance(content, list) and len(content) > 0 and getattr(content[0], "type", None) == "text" ): error_text = content[0].text else: error_text = json.dumps(content, default=str) raise RuntimeError(f'MCP tool "{name}" returned error: {error_text}') structured = getattr(result, "structuredContent", None) if structured is not None: return structured content = getattr(result, "content", None) if isinstance(content, list) and len(content) > 0: first = content[0] if getattr(first, "type", None) == "text": try: return json.loads(first.text) except (ValueError, TypeError): return first.text return content