#!/usr/bin/env python3 """seecrime.com MCP stdio bridge. Lets an MCP client that only launches LOCAL stdio servers (Claude Desktop, Cursor, VS Code, Gemini CLI, Windsurf and similar) use the remote seecrime.com MCP server. The client starts this script, writes JSON-RPC messages to its stdin (one per line) and reads the replies from its stdout. Each message is POSTed to the remote Streamable HTTP endpoint and every message the server sends back is written to stdout, one per line. Standard library only; Python 3.9 or newer. Run with --help for the configuration and a ready-to-paste Claude Desktop entry. Spec followed (both eras; the server supports 2026-07-28, 2025-11-25, 2025-06-18 and 2025-03-26): https://modelcontextprotocol.io/specification/2026-07-28/basic/transports/stdio https://modelcontextprotocol.io/specification/2026-07-28/basic/transports/streamable-http https://modelcontextprotocol.io/specification/2025-11-25/basic/transports """ import _socket import argparse import base64 import concurrent.futures import http.client import json import os import queue import re import socket import ssl import sys import threading import time import traceback import urllib.parse import urllib.request __version__ = '1.0.0' DEFAULT_URL = 'https://www.seecrime.com/mcp' DEFAULT_TIMEOUT = 60.0 MAX_TIMEOUT = 3600.0 # A reply from the server is read into memory before it is written to stdout, so # it is capped. For an SSE stream the cap applies to each event (a long-lived # stream may carry any number of them). MAX_RESPONSE_BYTES = 10 * 1024 * 1024 MAX_INPUT_LINE_BYTES = 10 * 1024 * 1024 # An error body is only read for a JSON-RPC error object or a short excerpt. MAX_ERROR_BODY_BYTES = 64 * 1024 MAX_EXCERPT_CHARS = 200 # Requests run on a small pool so that one slow tools/call does not hold up the # rest; notifications and client responses are sent in order on the dispatcher. WORKER_THREADS = 4 # Plain http:// is allowed only to these hosts (local development). LOOPBACK_HOSTS = frozenset(('localhost', '127.0.0.1', '::1')) # Legacy (2025-11-25) SSE resumption: how many reconnects in a row may deliver no # event before the bridge gives up. MAX_IDLE_RECONNECTS = 3 DEFAULT_RECONNECT_MS = 1000 # JSON-RPC 2.0 codes. PARSE_ERROR = -32700 INVALID_REQUEST = -32600 # Codes for errors the BRIDGE raises (the server never sent them). MCP 2026-07-28 # (basic/index, "Error Codes") reserves -32020..-32099 for codes the spec defines, # marks -32000..-32019 as legacy that new implementations SHOULD NOT use, and says # new codes SHOULD sit outside -32768..-32000 and that local errors must not be # mistakable for the peer's. So these use -31001.., and every one carries # data.source == "seecrime-mcp-bridge". ERR_AUTH = -31001 ERR_FORBIDDEN = -31002 ERR_NOT_FOUND = -31003 ERR_RATE_LIMITED = -31004 ERR_SERVER = -31005 ERR_HTTP = -31006 ERR_REDIRECT = -31007 ERR_UNREACHABLE = -31008 ERR_TIMEOUT = -31009 ERR_TLS = -31010 ERR_BAD_RESPONSE = -31011 ERR_TOO_LARGE = -31012 ERR_SESSION = -31013 ERR_INTERNAL = -31014 ERROR_SOURCE = 'seecrime-mcp-bridge' # 2026-07-28 HeaderMismatch: the tools/call retry below keys on it. HEADER_MISMATCH = -32020 META_VERSION_KEY = 'io.modelcontextprotocol/protocolVersion' # A Bearer credential is a token68 (RFC 6750 section 2.1). Anything else would # need escaping inside the Authorization header, i.e. it is not a valid key. _TOKEN68_RE = re.compile(r'^[A-Za-z0-9\-._~+/]+=*$') # Any seecrime API key, not just ours: redacted from everything the bridge writes. _ANY_KEY_RE = re.compile(r'sck_[0-9A-Fa-f]{12}_[A-Za-z0-9_\-]{16,}') # HTTP field-name token (RFC 9110 5.6.2): the shape an x-mcp-header name must have. _HEADER_TOKEN_RE = re.compile(r"^[!#$%&'*+\-.^_`|~0-9A-Za-z]+$") _DIGITS_RE = re.compile(r'^[0-9]+$') _EOL_RE = re.compile(rb'\r\n?|\n') _EOF = object() _STDOUT_GONE = object() _OVERSIZED = object() # --------------------------------------------------------------------------- helpers def _id_key(value): """A dictionary key for a JSON-RPC id. 1 and "1" are different ids.""" return json.dumps(value, sort_keys=True) def _reject_constant(name): raise ValueError('non-finite number %s is not JSON' % name) def _finite_float(text): value = float(text) if value != value or value in (float('inf'), float('-inf')): raise ValueError('number out of range') return value def loads_strict(text): """json.loads that refuses NaN / Infinity, which are not JSON (RFC 8259) and which json.dumps(allow_nan=False) could not write back out.""" return json.loads(text, parse_constant=_reject_constant, parse_float=_finite_float) def dumps_line(obj): """One JSON-RPC message as one line: compact and ASCII-only, so no raw newline (or U+2028) can ever split it (stdio spec: no embedded newlines).""" return json.dumps(obj, ensure_ascii=True, separators=(',', ':'), allow_nan=False) def _header_safe(value): """True for a non-empty value of visible ASCII only (no spaces, no controls).""" return isinstance(value, str) and bool(value) and all(0x21 <= ord(c) <= 0x7e for c in value) def encode_header_value(value): """2026-07-28 Streamable HTTP "Value Encoding" for Mcp-Name / Mcp-Param-*. A value that is not a plain header-safe ASCII string -- non-ASCII, a control character, leading or trailing whitespace -- is sent as =?base64??=, and so is any plain value that already looks like that sentinel, so the server can never mis-decode it. """ plain = all(c == '\t' or 0x20 <= ord(c) <= 0x7e for c in value) if plain and value == value.strip(' \t') and not ( value.startswith('=?base64?') and value.endswith('?=')): return value return '=?base64?' + base64.b64encode(value.encode('utf-8')).decode('ascii') + '?=' def _meta_protocol_version(msg): """The 2026-07-28 per-request protocol version, if the message carries one. Its presence is what marks a message as "modern" (stateless, per-request metadata); a legacy message never has it. """ params = msg.get('params') if isinstance(params, dict): meta = params.get('_meta') if isinstance(meta, dict): version = meta.get(META_VERSION_KEY) if isinstance(version, str): return version return None def _media_type(value): return (value or '').split(';', 1)[0].strip().lower() def _classify(msg): if not isinstance(msg, dict): return 'invalid' if 'method' in msg: if not isinstance(msg.get('method'), str): return 'invalid' return 'request' if 'id' in msg else 'notification' if 'result' in msg or 'error' in msg: return 'response' return 'invalid' def _usable_id(msg): """The id to answer an invalid message with: its own if it is a JSON-RPC id.""" if isinstance(msg, dict): value = msg.get('id') if isinstance(value, str) or (isinstance(value, int) and not isinstance(value, bool)): return value return None def _safe_label(value, limit=80): """A server- or client-supplied string made fit for one log line.""" text = value if isinstance(value, str) else repr(value) text = re.sub(r'[^\x20-\x7e]', '?', text) if len(text) > limit: text = text[:limit - 3] + '...' return text # --------------------------------------------------------------- x-mcp-header (2026) _SCHEMA_MAP_KEYWORDS = frozenset(('properties', 'patternProperties', '$defs', 'definitions', 'dependentSchemas')) _SCHEMA_DATA_KEYWORDS = frozenset(('enum', 'const', 'default', 'examples')) _HEADER_PARAM_TYPES = frozenset(('string', 'integer', 'boolean')) def _count_annotations(node): """Every x-mcp-header annotation anywhere in a schema (data keywords skipped, property NAMES never counted -- a property may be called "x-mcp-header").""" total = 0 if isinstance(node, dict): for key, value in node.items(): if key == 'x-mcp-header': total += 1 elif key in _SCHEMA_DATA_KEYWORDS: continue elif key in _SCHEMA_MAP_KEYWORDS and isinstance(value, dict): for sub in value.values(): total += _count_annotations(sub) else: total += _count_annotations(value) elif isinstance(node, list): for item in node: total += _count_annotations(item) return total def _primitive_type(schema): kind = schema.get('type') if isinstance(kind, str): return kind in _HEADER_PARAM_TYPES if isinstance(kind, list): rest = [k for k in kind if k != 'null'] return len(rest) == 1 and rest[0] in _HEADER_PARAM_TYPES return False def header_annotations(schema): """Validate a tool inputSchema's x-mcp-header annotations (2026-07-28 Streamable HTTP, "Schema Extension"). Returns (annotations, problem): annotations is a list of (property path, header name); problem is None or the reason the tool must be excluded. """ if not isinstance(schema, dict): return [], None found = [] def walk(node, path): props = node.get('properties') if not isinstance(props, dict): return for name, sub in props.items(): if not isinstance(sub, dict): continue if 'x-mcp-header' in sub: found.append((path + (name,), sub)) walk(sub, path + (name,)) walk(schema, ()) if _count_annotations(schema) != len(found): return None, 'an x-mcp-header annotation is not on a statically reachable property' seen = set() annotations = [] for path, sub in found: header = sub.get('x-mcp-header') if not isinstance(header, str) or not _HEADER_TOKEN_RE.match(header): return None, 'an x-mcp-header value is not a valid HTTP header name' if header.lower() in seen: return None, 'two x-mcp-header values are the same' seen.add(header.lower()) if not _primitive_type(sub): return None, 'an x-mcp-header is on a parameter that is not a string, integer or boolean' annotations.append((path, header)) return annotations, None def _value_at(arguments, path): node = arguments for step in path: if not isinstance(node, dict) or step not in node: return False, None node = node[step] return True, node def _param_text(value): """Type conversion for an Mcp-Param-* header (2026-07-28 "Value Encoding").""" if isinstance(value, bool): return 'true' if value else 'false' if isinstance(value, int): return str(value) if isinstance(value, float): return str(int(value)) if value.is_integer() else repr(value) if isinstance(value, str): return value return None # ----------------------------------------------------------------- configuration class ConfigError(Exception): """A configuration problem: reported as one line on stderr, exit status 2.""" class Proxy(object): def __init__(self, host, port, headers, secret): self.host = host self.port = port self.headers = headers self.secret = secret class Config(object): def __init__(self, api_key, url, timeout, verbose=False, proxy=None): self.api_key = api_key self.timeout = timeout self.verbose = verbose self.scheme, self.host, self.port, self.target = parse_endpoint(url) self.proxy = proxy @property def display_endpoint(self): """The endpoint for a log line: scheme, host, port and path, no query.""" host = '[%s]' % self.host if ':' in self.host else self.host default = 443 if self.scheme == 'https' else 80 port = '' if self.port == default else ':%d' % self.port return '%s://%s%s%s' % (self.scheme, host, port, self.target.split('?', 1)[0]) def parse_endpoint(url): """Validate SEECRIME_MCP_URL; return (scheme, host, port, request target). https:// is required, except plain http:// to localhost / 127.0.0.1 / [::1] for local development: anywhere else it would send the API key in clear. A user name or password in the URL is refused -- no credential travels in a URL; the key goes in the Authorization header only. """ if not isinstance(url, str) or not url.strip(): raise ConfigError('SEECRIME_MCP_URL is empty') try: parts = urllib.parse.urlsplit(url.strip()) port = parts.port except ValueError: raise ConfigError('SEECRIME_MCP_URL is not a valid URL') scheme = parts.scheme.lower() if scheme not in ('https', 'http'): raise ConfigError('SEECRIME_MCP_URL must start with https://') if '@' in parts.netloc: raise ConfigError('SEECRIME_MCP_URL must not contain a user name or password; ' 'the API key belongs in SEECRIME_API_KEY') host = parts.hostname if not host: raise ConfigError('SEECRIME_MCP_URL has no host name') if scheme == 'http' and host not in LOOPBACK_HOSTS: raise ConfigError('SEECRIME_MCP_URL uses http:// for a host that is not local; ' 'use https:// (plain http:// is allowed only for localhost, ' '127.0.0.1 and [::1])') target = parts.path or '/' if parts.query: target += '?' + parts.query if any(not 0x21 <= ord(c) <= 0x7e for c in target): raise ConfigError('SEECRIME_MCP_URL contains characters that are not allowed in a URL') if port is None: port = 443 if scheme == 'https' else 80 return scheme, host, port, target def parse_proxy(value): """An HTTPS proxy setting -> Proxy, or None for one this bridge cannot use.""" text = value.strip() if '://' not in text: text = 'http://' + text try: parts = urllib.parse.urlsplit(text) port = parts.port except ValueError: raise ConfigError('the HTTPS proxy setting (HTTPS_PROXY) is not a valid URL') if parts.scheme.lower() not in ('http', 'https') or not parts.hostname: return None headers = {} secret = None if parts.username is not None: user = urllib.parse.unquote(parts.username) secret = urllib.parse.unquote(parts.password or '') token = base64.b64encode((user + ':' + secret).encode('utf-8')).decode('ascii') headers['Proxy-Authorization'] = 'Basic ' + token if port is None: port = 443 if parts.scheme.lower() == 'https' else 80 return Proxy(parts.hostname, port, headers, secret) def find_proxy(scheme, host, log=None): """The HTTPS proxy from the environment (or, on Windows, the system settings), the same sources urllib uses. Loopback is never proxied.""" if scheme != 'https' or host in LOOPBACK_HOSTS: return None try: proxies = urllib.request.getproxies() except Exception: return None value = proxies.get('https') or proxies.get('all') if not value: return None try: if urllib.request.proxy_bypass(host): return None except Exception: pass proxy = parse_proxy(value) if proxy is None and log is not None: log.warn('ignoring the proxy setting: only http:// (CONNECT) proxies are supported') return proxy def _env(environ, name): """An environment value; empty or whitespace means unset.""" value = environ.get(name) if value is None or not value.strip(): return None return value.strip() def load_config(args, environ): api_key = args.api_key if args.api_key else _env(environ, 'SEECRIME_API_KEY') if not api_key or not api_key.strip(): raise ConfigError('SEECRIME_API_KEY is not set. Create an API key at ' 'https://www.seecrime.com/developers and put it in the "env" ' 'block of your MCP client configuration (run with --help for ' 'an example).') api_key = api_key.strip() if not _TOKEN68_RE.match(api_key) or len(api_key) < 8: raise ConfigError('SEECRIME_API_KEY does not look like an API key (it contains ' 'spaces or other characters a key never has, or it is too short)') url = args.url if args.url else (_env(environ, 'SEECRIME_MCP_URL') or DEFAULT_URL) raw_timeout = args.timeout if args.timeout else _env(environ, 'SEECRIME_MCP_TIMEOUT') timeout = DEFAULT_TIMEOUT if raw_timeout is not None: try: timeout = float(raw_timeout) except ValueError: timeout = -1.0 if not (0 < timeout <= MAX_TIMEOUT): raise ConfigError('SEECRIME_MCP_TIMEOUT must be a number of seconds between ' '1 and %d' % MAX_TIMEOUT) return Config(api_key, url, timeout, verbose=bool(args.verbose)) # ------------------------------------------------------------- logging and output class Redactor(object): """Removes the API key (and any proxy password) from everything written out.""" def __init__(self, secrets): forms = set() for secret in secrets: if secret: forms.add(secret) forms.add(urllib.parse.quote(secret, safe='')) self._forms = sorted((f for f in forms if f), key=len, reverse=True) def text(self, value): for form in self._forms: value = value.replace(form, '[REDACTED]') return _ANY_KEY_RE.sub('[REDACTED]', value) def obj(self, value): """Redact inside the strings of a JSON value, never across its syntax.""" if isinstance(value, str): return self.text(value) if isinstance(value, dict): return {self.text(k): self.obj(v) for k, v in value.items()} if isinstance(value, list): return [self.obj(v) for v in value] return value class Log(object): """Diagnostics on stderr: one redacted line each, never on stdout.""" def __init__(self, redactor, verbose=False, stream=None): self._redactor = redactor self.verbose = verbose self._stream = stream self._lock = threading.Lock() def _write(self, text): line = 'seecrime-mcp: ' + self._redactor.text(text) line = ''.join(c if (c >= ' ' or c == '\t') else ' ' for c in line) stream = self._stream if self._stream is not None else sys.stderr try: with self._lock: target = getattr(stream, 'buffer', None) if target is not None: target.write(line.encode('utf-8', 'replace') + b'\n') target.flush() else: stream.write(line + '\n') stream.flush() except Exception: pass def info(self, text): self._write(text) warn = info error = info def debug(self, text): if self.verbose: self._write(text) class Output(object): """stdout: valid MCP messages only, one per line, flushed (stdio spec).""" def __init__(self, stream, redactor, on_broken): self._stream = stream self._redactor = redactor self._on_broken = on_broken self._lock = threading.Lock() self.broken = False def write(self, obj): """Write one message (or one batch array). Returns False if stdout is gone; raises Unwritable for a value that cannot be written as JSON.""" try: data = dumps_line(self._redactor.obj(obj)).encode('ascii') + b'\n' except (TypeError, ValueError, RecursionError): raise Unwritable() with self._lock: if self.broken: return False try: self._stream.write(data) self._stream.flush() except (OSError, ValueError): self.broken = True if self.broken: self._on_broken() return False return True def hold(self, timeout): """Take the write lock so no line is cut short by exit; bounded.""" return self._lock.acquire(timeout=timeout) # -------------------------------------------------------------------- errors class BridgeError(Exception): def __init__(self, code, message, **data): Exception.__init__(self, message) self.code = code self.message = message self.data = data class Cancelled(Exception): """The client cancelled the request (or the bridge is shutting down).""" class Unwritable(Exception): """A value json cannot write (nested too deeply, for one).""" def _error_message(msg_id, code, message): """A plain JSON-RPC error response (for input the bridge itself rejects).""" return {'jsonrpc': '2.0', 'id': msg_id, 'error': {'code': code, 'message': message}} def bridge_error_object(code, message, **data): payload = {'source': ERROR_SOURCE} for key, value in data.items(): if value is not None: payload[key] = value return {'code': code, 'message': message, 'data': payload} # --------------------------------------------------------------------- calls def _socket_of(conn): """The socket under a connection, also once a response owns it. For an HTTP/1.0 or `Connection: close` response, http.client's getresponse() hands the socket to the response and sets conn.sock to None, so the socket is recorded when the connection is made (see Bridge._open). """ return getattr(conn, '_seecrime_sock', None) or getattr(conn, 'sock', None) class Call(object): """One JSON-RPC request in flight: its connection (so a cancellation can close the stream) and whether anything more may be written for it.""" def __init__(self, msg, raw, internal=False, parent=None): self.msg = msg self.raw = raw self.id = msg.get('id') self.key = _id_key(self.id) self.method = msg.get('method') self.version = _meta_protocol_version(msg) self.modern = self.version is not None self.internal = internal self.collected = [] self.batch = None # the Batch this request arrived in, if any self._owner = parent if parent is not None else self self._lock = threading.Lock() self._conns = set() self._cancelled = False self._started = False self._event = threading.Event() @property def cancelled(self): return self._owner._cancelled @property def cancel_event(self): return self._owner._event def attach(self, conn): owner = self._owner with owner._lock: if owner._cancelled: return False owner._conns.add(conn) owner._started = True return True def detach(self, conn): owner = self._owner with owner._lock: owner._conns.discard(conn) def cancel(self): """Mark cancelled; return whether its HTTP request had been started.""" owner = self._owner with owner._lock: started = owner._started owner._cancelled = True owner._event.set() return started def abort(self): """Close the connection under the request. On 2026-07-28 Streamable HTTP closing the response stream IS the cancellation signal. Both steps are needed to free the worker blocked reading it: - shutdown() sends the FIN and, on Linux, wakes the reader; - on Windows it does not wake a recv/select while the server is quiet (a WSGI worker still computing), but closing the socket does. The close is the C-level one: socket.close() defers the real close while http.client's response holds a makefile reference, i.e. exactly while a read is in progress. The C close invalidates the object's descriptor first, so the woken reader fails cleanly (and is mapped to Cancelled). """ owner = self._owner with owner._lock: conns = list(owner._conns) for conn in conns: sock = _socket_of(conn) if sock is None: continue try: # The plain-socket shutdown, also for a TLS socket: no SSL state. socket.socket.shutdown(sock, socket.SHUT_RDWR) except (OSError, ValueError): pass try: _socket.socket.close(sock) except (OSError, ValueError): pass class Batch(object): """The answers to one stdin batch (2025-03-26: implementations MUST support receiving JSON-RPC batches), written as ONE array line once every request in it has been answered -- JSON-RPC 2.0 batch semantics. A notification gets no answer; a cancelled request is left out; a batch with no answers writes nothing. Server notifications on a member's stream still go out as they arrive: they are not answers.""" def __init__(self, write): self._write = write self._lock = threading.Lock() self._pending = set() self._answers = [] self._sealed = False self._done = False def expect(self, call): with self._lock: self._pending.add(call) def answer(self, msg, call=None): with self._lock: self._answers.append(msg) self._pending.discard(call) self._flush_if_complete() def drop(self, call): with self._lock: if call in self._pending: self._pending.discard(call) self._flush_if_complete() def seal(self): """Every element has been dispatched; nothing more will be expected.""" with self._lock: self._sealed = True self._flush_if_complete() def _flush_if_complete(self): if self._sealed and not self._pending and not self._done: self._done = True if self._answers: self._write(list(self._answers)) # ------------------------------------------------------------------ SSE parser class SSEStream(object): """text/event-stream parser per the WHATWG "event stream interpretation": CRLF, LF and lone CR line ends; ':' comments; data / event / id / retry fields; an event is dispatched at a blank line, and an unterminated event at end of stream is discarded.""" def __init__(self, read, max_event_bytes=MAX_RESPONSE_BYTES): self._read = read self._max = max_event_bytes self._id_buffer = '' self.last_event_id = '' self.retry_ms = None self.events_seen = 0 def _lines(self): buf = bytearray() pos = 0 # start of the line being assembled scan = 0 # no line end lies before this offset first = True while True: while True: match = _EOL_RE.search(buf, scan) if match is None: scan = len(buf) break if match.end() == len(buf) and buf[-1] == 0x0D: scan = match.start() # a CR at the end: wait to see if LF follows break line = bytes(buf[pos:match.start()]) pos = scan = match.end() yield line.decode('utf-8', 'replace') if pos: del buf[:pos] scan -= pos pos = 0 if len(buf) > self._max: raise BridgeError(ERR_TOO_LARGE, 'seecrime MCP: an event from the server is ' 'larger than the bridge accepts (%d MB)' % (self._max // (1024 * 1024))) chunk = self._read() if not chunk: if buf[-1:] == b'\r': yield bytes(buf[:-1]).decode('utf-8', 'replace') return # an unterminated last line is discarded if first: first = False if chunk.startswith(b'\xef\xbb\xbf'): chunk = chunk[3:] buf.extend(chunk) def events(self): """Yield the data of each dispatched message event, as text.""" data = [] size = 0 event_type = '' for line in self._lines(): if not line: self.last_event_id = self._id_buffer self.events_seen += 1 if data: payload = '\n'.join(data) kind = event_type data = [] size = 0 event_type = '' if kind in ('', 'message'): yield payload else: event_type = '' continue if line.startswith(':'): continue field, sep, value = line.partition(':') if sep and value.startswith(' '): value = value[1:] if field == 'data': size += len(value) + 1 if size > self._max: raise BridgeError(ERR_TOO_LARGE, 'seecrime MCP: an event from the server ' 'is larger than the bridge accepts (%d MB)' % (self._max // (1024 * 1024))) data.append(value) elif field == 'event': event_type = value elif field == 'id': if '\x00' not in value: self._id_buffer = value elif field == 'retry': if _DIGITS_RE.match(value): self.retry_ms = int(value) def _close(call, conn, resp): """Release an exchange now. Closing the connection alone is not enough: for an HTTP/1.0 or `Connection: close` response the socket belongs to the response, and would otherwise stay open until garbage collection.""" if call is not None: call.detach(conn) for closable in (resp, conn): try: closable.close() except Exception: pass def _reader_for(resp): read1 = getattr(resp, 'read1', None) if read1 is not None: return lambda: read1(65536) return lambda: resp.readline(65536) def _parse_messages(text): """A JSON body or SSE data -> list of message objects (an array is split).""" value = loads_strict(text) if isinstance(value, dict): return [value] if isinstance(value, list) and value and all(isinstance(v, dict) for v in value): return list(value) raise ValueError('not a JSON-RPC message') def _tls_context(): """Certificate and host-name verification always on (the default context); there is deliberately no way to turn it off.""" ctx = ssl.create_default_context() ctx.minimum_version = ssl.TLSVersion.TLSv1_2 return ctx # ------------------------------------------------------------------------ bridge class Bridge(object): def __init__(self, cfg, stdin=None, stdout=None, stderr=None): self.cfg = cfg # stdout carries the server's data, so only what could have reached the # server is redacted there: the API key (and any sck_ key). The proxy # password never reaches it, and redacting a short one ("2026") out of # replies would corrupt them while protecting nothing. The log, which # holds the bridge's own text, redacts both. self.redactor = Redactor([cfg.api_key]) self.log = Log(Redactor([cfg.api_key, cfg.proxy.secret if cfg.proxy else None]), cfg.verbose, stream=stderr) self._stdin = stdin self.out = Output(stdout, self.redactor, self._stdout_gone) self._ssl = _tls_context() if cfg.scheme == 'https' else None self._user_agent = 'seecrime-mcp-stdio/%s (Python %d.%d)' % ( __version__, sys.version_info[0], sys.version_info[1]) self.inbox = queue.Queue() self._pool = concurrent.futures.ThreadPoolExecutor( max_workers=WORKER_THREADS, thread_name_prefix='seecrime-mcp') self._futures = set() self._calls = {} self._calls_lock = threading.Lock() # Legacy (initialize-based) session state. self._state_lock = threading.Lock() self._reinit_lock = threading.Lock() self.session_id = None self.negotiated_version = None self._init_params = None # 2026-07-28 x-mcp-header annotations, from tools/list replies. self._tools_lock = threading.Lock() self._tool_headers = {} self._internal_ids = 0 # ------------------------------------------------------------ run loop def run(self): via = (' via proxy %s:%d' % (self.cfg.proxy.host, self.cfg.proxy.port) if self.cfg.proxy else '') self.log.info('bridge %s forwarding stdio to %s%s' % (__version__, self.cfg.display_endpoint, via)) reader = threading.Thread(target=self._read_stdin, name='seecrime-mcp-stdin') reader.daemon = True reader.start() drain = True try: while True: try: item = self.inbox.get(timeout=0.5) except queue.Empty: continue if item is _EOF: break if item is _STDOUT_GONE: drain = False break try: self._dispatch(item) except Exception as exc: self.log.error('internal error while handling a message (%s)' % type(exc).__name__) self._debug_traceback() except KeyboardInterrupt: self._shutdown(drain=False) self.out.hold(2.0) return 130 self._shutdown(drain=drain) self.out.hold(2.0) # so the exit that follows cannot cut a line short return 0 def _read_stdin(self): """Drain stdin into the inbox, so a client that pipelines requests and does not read stdout until it has written them all can never deadlock against the bridge's own writes.""" stream = self._stdin try: while True: line = stream.readline(MAX_INPUT_LINE_BYTES + 1) if not line: break if len(line) > MAX_INPUT_LINE_BYTES and not line.endswith(b'\n'): while True: rest = stream.readline(65536) if not rest or rest.endswith(b'\n'): break self.inbox.put(_OVERSIZED) continue self.inbox.put(line) except Exception as exc: self.log.error('stopped reading stdin (%s)' % type(exc).__name__) finally: self.inbox.put(_EOF) def _stdout_gone(self): self.inbox.put(_STDOUT_GONE) # ------------------------------------------------------------ dispatch def _dispatch(self, item): if item is _OVERSIZED: self.log.warn('a message on stdin is larger than %d MB; rejected' % (MAX_INPUT_LINE_BYTES // (1024 * 1024))) self._write_error(None, INVALID_REQUEST, 'Invalid Request: message too large') return line = item.strip() if not line: return if line.startswith(b'\xef\xbb\xbf'): line = line[3:].strip() try: value = loads_strict(line.decode('utf-8')) except (ValueError, RecursionError): # UnicodeDecodeError and JSONDecodeError are ValueErrors; JSON nested # deeper than the interpreter's recursion limit is a RecursionError. self.log.warn('a line on stdin is not valid JSON; answered with a parse error') self._write_error(None, PARSE_ERROR, 'Parse error') return if isinstance(value, list): self._dispatch_batch(value) return self._dispatch_message(value, line) def _dispatch_batch(self, elements): """A 2025-03-26 batch: each element is forwarded on its own, and the answers go back as one array line (see Batch).""" if not elements: self._write_error(None, INVALID_REQUEST, 'Invalid Request: empty batch') return batch = Batch(self.out.write) for element in elements: raw = None if isinstance(element, dict): try: raw = dumps_line(element).encode('ascii') except (ValueError, RecursionError): raw = None if raw is None: batch.answer(_error_message(None, INVALID_REQUEST, 'Invalid Request')) continue self._dispatch_message(element, raw, batch) batch.seal() def _dispatch_message(self, msg, raw, batch=None): kind = _classify(msg) if kind == 'invalid': error = _error_message(_usable_id(msg), INVALID_REQUEST, 'Invalid Request') if batch is not None: batch.answer(error) else: self.out.write(error) return if kind == 'request': call = Call(msg, raw) if batch is not None: call.batch = batch batch.expect(call) if call.method == 'initialize': # Inline: the session and negotiated version it sets must be in # place before any later message is forwarded. self._handle_request(call, register=False) return with self._calls_lock: self._calls[call.key] = call if call.method == 'subscriptions/listen': # A subscription stream stays open until it is cancelled, so it # gets its own thread: it must never hold one of the pool's # workers, or a few subscriptions would stall every request. thread = threading.Thread(target=self._handle_request, args=(call,), name='seecrime-mcp-listen') thread.daemon = True thread.start() return future = self._pool.submit(self._handle_request, call) self._futures.add(future) future.add_done_callback(self._futures.discard) return if kind == 'notification' and msg.get('method') == 'notifications/cancelled': self._handle_cancel(msg, raw) return # Notifications and the client's responses (to a legacy server's # request) go out in order, before anything that follows them. self._forward_oneway(msg, raw) # ------------------------------------------------------------ requests def _handle_request(self, call, register=True): try: if call.cancelled: return self._forward_request(call) except Cancelled: pass except BridgeError as exc: self._emit_error(call, exc.code, exc.message, **exc.data) if exc.code == ERR_TIMEOUT and not call.modern and call.method != 'initialize': # 2025-11-25 lifecycle: initialize MUST NOT be cancelled. self._cancel_upstream(call) except Exception as exc: self.log.error('internal error while forwarding %s (%s)' % (_safe_label(call.method), type(exc).__name__)) self._debug_traceback() self._emit_error(call, ERR_INTERNAL, 'seecrime MCP: internal error in the stdio ' 'bridge (see its log)') finally: if register: with self._calls_lock: if self._calls.get(call.key) is call: del self._calls[call.key] if call.batch is not None: call.batch.drop(call) # unanswered (cancelled): the array omits it def _forward_request(self, call): retried_session = False refreshed_tools = False while True: headers, session_used = self._headers_for(call.msg, 'request') started = time.monotonic() try: outcome = self._post(call, headers) except BridgeError as exc: if call.cancelled: raise Cancelled() self.log.error('%s failed: %s' % (_safe_label(call.method), exc.message)) raise self.log.debug('%s -> %s (%d ms)' % ( _safe_label(call.method), 'ok' if outcome is None else 'HTTP %d' % outcome[0], (time.monotonic() - started) * 1000)) if outcome is None: return status, ctype, body, resp_headers = outcome if (status == 404 and session_used and not retried_session and call.method != 'initialize'): retried_session = True if self._reestablish_session(session_used, call): continue raise BridgeError(ERR_SESSION, 'seecrime MCP: the server ended the session and ' 'a new one could not be started - restart the MCP server ' 'in your client', httpStatus=status) if (status == 400 and call.modern and call.method == 'tools/call' and not refreshed_tools and not call.internal and self._rpc_code(call, body) == HEADER_MISMATCH): # 2026-07-28: a HeaderMismatch on tools/call usually means the # tool's x-mcp-header annotations changed or were never seen; # re-read tools/list and retry once. refreshed_tools = True if self._refresh_tools(call): continue self._emit_status_error(call, status, ctype, body, resp_headers) return def _post(self, call, headers): """POST the message; relay what the server sends for it. Returns None once the final response has been handled, or (status, media type, body, headers) for a non-2xx answer. """ conn, resp = self._open(call, 'POST', headers, call.raw) try: status = resp.status ctype = _media_type(resp.getheader('Content-Type')) if 200 <= status < 300: if status in (202, 204): raise BridgeError(ERR_BAD_RESPONSE, 'seecrime MCP: the server accepted the ' 'request but sent no response', httpStatus=status) if ctype == 'text/event-stream': self._relay_sse(call, conn, resp) return None if ctype == 'application/json': body = self._read_body(call, resp, MAX_RESPONSE_BYTES, truncate=False) self._relay_json(call, body, resp) return None raise BridgeError(ERR_BAD_RESPONSE, 'seecrime MCP: the server answered with an ' 'unexpected content type', httpStatus=status) body = self._read_body(call, resp, MAX_ERROR_BODY_BYTES, truncate=True) return status, ctype, body, resp finally: _close(call, conn, resp) def _relay_json(self, call, body, resp): try: messages = _parse_messages(body.decode('utf-8-sig')) except (ValueError, RecursionError): raise BridgeError(ERR_BAD_RESPONSE, 'seecrime MCP: the server sent a reply that ' 'is not valid JSON-RPC') final = None for msg in messages: if final is None and self._is_final(call, msg): final = msg else: self._relay_other(call, msg) if final is None: raise BridgeError(ERR_BAD_RESPONSE, 'seecrime MCP: the server\'s reply did not ' 'answer this request') self._deliver_final(call, final, resp) def _relay_sse(self, call, conn, resp): """Relay an SSE response stream until the final response for `call`. Notifications (and, from a legacy server, requests) on the stream are written to stdout as they arrive. A legacy (2025-11-25) stream that ends before the response is resumed with GET + Last-Event-ID, honouring `retry`; 2026-07-28 streams are not resumable. """ if call.method == 'subscriptions/listen' and _socket_of(conn) is not None: # A listen stream may legitimately stay quiet longer than the # timeout. It is not given a finite timeout in a loop instead: once # a socket read times out, http.client's reader refuses every later # read, so the stream could not go on. It runs on its own thread # (it never holds a pool worker), and abort() closes the socket, # which wakes it when the client cancels and at shutdown. _socket_of(conn).settimeout(None) parser = SSEStream(_reader_for(resp)) idle_reconnects = 0 current = (conn, resp) try: while True: before = parser.events_seen try: for data in parser.events(): if call.cancelled: raise Cancelled() if not data.strip(): continue # e.g. a priming event: an id and empty data try: messages = _parse_messages(data) except (ValueError, RecursionError): self.log.warn('ignored an SSE event that is not a JSON-RPC message') continue for msg in messages: if self._is_final(call, msg): self._deliver_final(call, msg, current[1]) return if (self._relay_other(call, msg) and self._server_cancelled(call, msg)): return # the server tore down this subscription stream except BridgeError: raise except (OSError, http.client.HTTPException, ValueError) as exc: if call.cancelled: raise Cancelled() if isinstance(exc, (socket.timeout, TimeoutError)): raise self._timeout_error() self.log.debug('the SSE stream broke (%s)' % type(exc).__name__) if call.cancelled: raise Cancelled() if call.modern or not parser.last_event_id: raise BridgeError(ERR_BAD_RESPONSE, 'seecrime MCP: the server closed the ' 'stream before it answered the request') idle_reconnects = 0 if parser.events_seen > before else idle_reconnects + 1 if idle_reconnects >= MAX_IDLE_RECONNECTS: raise BridgeError(ERR_BAD_RESPONSE, 'seecrime MCP: the server kept closing ' 'the stream without answering the request') current = self._resume(call, parser, current) parser._read = _reader_for(current[1]) finally: if current[0] is not conn: _close(call, *current) def _resume(self, call, parser, current): """Reconnect a legacy SSE stream: GET with Last-Event-ID (2025-11-25).""" _close(call, *current) delay = (parser.retry_ms if parser.retry_ms is not None else DEFAULT_RECONNECT_MS) / 1000.0 if delay > self.cfg.timeout: raise BridgeError(ERR_BAD_RESPONSE, 'seecrime MCP: the server asked the bridge to ' 'reconnect later than SEECRIME_MCP_TIMEOUT allows') if call.cancel_event.wait(delay): raise Cancelled() if not _header_safe(parser.last_event_id): raise BridgeError(ERR_BAD_RESPONSE, 'seecrime MCP: the stream cannot be resumed') headers = self._base_headers(accept='text/event-stream') with self._state_lock: if self.negotiated_version: headers['MCP-Protocol-Version'] = self.negotiated_version if self.session_id: headers['Mcp-Session-Id'] = self.session_id headers['Last-Event-ID'] = parser.last_event_id self.log.debug('resuming the SSE stream for %s' % _safe_label(call.method)) conn, resp = self._open(call, 'GET', headers, None) if resp.status != 200 or _media_type(resp.getheader('Content-Type')) != 'text/event-stream': status = resp.status _close(call, conn, resp) raise BridgeError(ERR_BAD_RESPONSE, 'seecrime MCP: the server closed the stream ' 'and could not resume it', httpStatus=status) return conn, resp @staticmethod def _server_cancelled(call, msg): """2026-07-28: a server ends a subscriptions/listen stream by sending notifications/cancelled for that request; nothing follows it.""" if msg.get('method') != 'notifications/cancelled': return False params = msg.get('params') return isinstance(params, dict) and _id_key(params.get('requestId')) == call.key @staticmethod def _is_final(call, msg): if msg.get('jsonrpc') != '2.0': return False if 'method' in msg or not ('result' in msg or 'error' in msg): return False if 'id' in msg and msg['id'] is not None: return _id_key(msg['id']) == call.key return 'error' in msg # an error with no id answers the one request in flight def _relay_other(self, call, msg): """A message on this exchange that is not its answer. Only a notification goes to stdout -- or, from a legacy server, a request (sampling, elicitation, roots). Anything else is dropped and logged: a response for ANOTHER id would be taken by the client as the answer to a request still in flight, and a 2026-07-28 server sends no requests at all (MRTR embeds them in results). Returns whether it was relayed.""" method = msg.get('method') if (msg.get('jsonrpc') == '2.0' and isinstance(method, str) and not (call.modern and 'id' in msg)): self._emit(call, msg) return True if 'result' in msg or 'error' in msg: what = 'a response that does not answer this request' elif msg.get('jsonrpc') == '2.0' and isinstance(method, str): what = 'a server request on a 2026-07-28 exchange' else: what = 'an object that is not a JSON-RPC message' self.log.warn('dropped %s, sent while the server answered %s' % (what, _safe_label(call.method))) return False def _deliver_final(self, call, msg, resp): if msg.get('id') is None: msg = dict(msg) msg['id'] = call.id if call.method == 'initialize' and isinstance(msg.get('result'), dict): self._capture_initialize(call, msg['result'], resp) elif (call.method == 'tools/list' and call.modern and isinstance(msg.get('result'), dict)): msg = self._filter_tools(msg) self._emit(call, msg, final=True) def _emit(self, call, msg, final=False): """Write a message for `call`. `final` marks its answer: an answer in a batch waits for the batch's array, and an answer that cannot be written as JSON is replaced by an error, so the client is never left waiting.""" if call.internal: call.collected.append(msg) return if call.cancelled: return # nothing more may be sent for a cancelled request if final and call.batch is not None: try: dumps_line(self.redactor.obj(msg)) except (TypeError, ValueError, RecursionError): msg = self._unwritable(call) call.batch.answer(msg, call) return try: self.out.write(msg) except Unwritable: self.log.warn('dropped a message from the server that cannot be written as JSON') if final: self.out.write(self._unwritable(call)) @staticmethod def _unwritable(call): return {'jsonrpc': '2.0', 'id': call.id, 'error': bridge_error_object( ERR_BAD_RESPONSE, 'seecrime MCP: the server\'s reply could not be relayed ' '(it is nested too deeply)')} def _emit_error(self, call, code, message, **data): self._emit(call, {'jsonrpc': '2.0', 'id': call.id, 'error': bridge_error_object(code, message, **data)}, final=True) def _write_error(self, msg_id, code, message): self.out.write(_error_message(msg_id, code, message)) # ------------------------------------------------------- error mapping @staticmethod def _rpc_error(body): """A JSON-RPC 2.0 error response in an error body, as (error object, id) -- id is None for a null or missing id -- or (None, None).""" try: value = loads_strict(body.decode('utf-8-sig')) except (ValueError, RecursionError): return None, None if (isinstance(value, dict) and value.get('jsonrpc') == '2.0' and 'method' not in value and isinstance(value.get('error'), dict)): err = value['error'] code = err.get('code') if (isinstance(code, int) and not isinstance(code, bool) and isinstance(err.get('message'), str)): out = {'code': code, 'message': err['message']} if 'data' in err: out['data'] = err['data'] return out, value.get('id') return None, None @staticmethod def _answers(call, rid): """Whether an error response with id `rid` answers `call`. The POST carried exactly ONE request, so an error with a null or missing id -- from a server that refused before it read the id (an auth, rate-limit, parse or size refusal) -- answers that request. A different id answers nothing.""" return rid is None or _id_key(rid) == call.key def _rpc_code(self, call, body): err, rid = self._rpc_error(body) return err['code'] if err is not None and self._answers(call, rid) else None def _excerpt(self, ctype, body): """A short, safe excerpt of an error body: redacted, printable ASCII, at most MAX_EXCERPT_CHARS, and only from a JSON or plain-text body (never from an HTML error page).""" text = None err, _ = self._rpc_error(body) if err is not None: text = err['message'] else: try: value = loads_strict(body.decode('utf-8-sig')) except (ValueError, RecursionError): value = None if isinstance(value, dict): for field in ('message', 'error', 'detail'): if isinstance(value.get(field), str): text = value[field] break elif ctype in ('text/plain', ''): text = body.decode('utf-8', 'replace') if not text: return None text = self.redactor.text(text) # before truncation: no key fragment survives text = ' '.join(re.sub(r'[^\x20-\x7e]+', ' ', text).split()) if len(text) > MAX_EXCERPT_CHARS: text = text[:MAX_EXCERPT_CHARS - 3] + '...' return text or None def _emit_status_error(self, call, status, ctype, body, resp): rpc, rid = self._rpc_error(body) label = _safe_label(call.method) detail = None if rpc is not None and not self._answers(call, rid): self.log.warn('dropped an error response for a different request (HTTP %d), sent ' 'while the server answered %s' % (status, label)) elif rpc is not None and not 300 <= status < 400: # The server's own JSON-RPC answer to this request, at any status: a # protocol error (UnsupportedProtocolVersion, HeaderMismatch, invalid # params, Method not found) that a client negotiates on, or a # refusal (401/403/429/5xx) whose code, message and data say why. # It passes through as sent, under the request's own id. self.log.info('%s: the server answered HTTP %d with JSON-RPC error %d' % (label, status, rpc['code'])) self._emit(call, {'jsonrpc': '2.0', 'id': call.id, 'error': rpc}, final=True) return elif rpc is None: detail = self._excerpt(ctype, body) # No JSON-RPC answer for this request (or a redirect): the bridge's own # error, so the client is never left waiting. if status == 401: code, message = ERR_AUTH, ('seecrime MCP: authentication failed (HTTP 401) - ' 'check SEECRIME_API_KEY') elif status == 403: code, message = ERR_FORBIDDEN, ('seecrime MCP: access denied (HTTP 403) - this ' 'API key may not use the MCP server') elif status == 429: code, message = ERR_RATE_LIMITED, ('seecrime MCP: rate limit reached (HTTP 429) ' '- wait a minute and try again') elif 300 <= status < 400: code, message = ERR_REDIRECT, ('seecrime MCP: the server answered with a ' 'redirect (HTTP %d), which the bridge does not ' 'follow - check SEECRIME_MCP_URL' % status) elif status >= 500: code, message = ERR_SERVER, ('seecrime MCP: the server failed to handle the ' 'request (HTTP %d) - try again shortly' % status) elif status == 404: code, message = ERR_NOT_FOUND, ('seecrime MCP: there is no MCP endpoint at ' 'SEECRIME_MCP_URL (HTTP 404)') else: code, message = ERR_HTTP, ('seecrime MCP: the server rejected the request ' '(HTTP %d)' % status) retry_after = resp.getheader('Retry-After') if status == 429 else None if retry_after is not None: retry_after = retry_after.strip() if not _header_safe(retry_after.replace(' ', '')) or len(retry_after) > 64: retry_after = None self.log.error('%s: %s' % (label, message)) self._emit_error(call, code, message, httpStatus=status, detail=detail, retryAfter=retry_after) def _timeout_error(self): return BridgeError(ERR_TIMEOUT, 'seecrime MCP: no answer from the server within %g ' 'seconds (SEECRIME_MCP_TIMEOUT)' % self.cfg.timeout) def _network_error(self, exc): host = self.cfg.host name = type(exc).__name__ if isinstance(exc, ssl.SSLCertVerificationError): return BridgeError(ERR_TLS, 'seecrime MCP: the TLS certificate of %s could not be ' 'verified' % host, error=name) if isinstance(exc, ssl.SSLError): return BridgeError(ERR_TLS, 'seecrime MCP: the TLS connection to %s failed' % host, error=name) if isinstance(exc, (socket.timeout, TimeoutError)): return self._timeout_error() if self.cfg.proxy is not None and 'Tunnel connection failed' in str(exc): return BridgeError(ERR_UNREACHABLE, 'seecrime MCP: the HTTPS proxy refused the ' 'connection to %s' % host, error=name) if isinstance(exc, ConnectionRefusedError): return BridgeError(ERR_UNREACHABLE, 'seecrime MCP: %s refused the connection - ' 'check SEECRIME_MCP_URL' % host, error=name) if isinstance(exc, socket.gaierror): return BridgeError(ERR_UNREACHABLE, 'seecrime MCP: could not look up %s - check ' 'your network connection' % host, error=name) if isinstance(exc, http.client.HTTPException): return BridgeError(ERR_UNREACHABLE, 'seecrime MCP: the connection to %s closed ' 'before the server answered' % host, error=name) return BridgeError(ERR_UNREACHABLE, 'seecrime MCP: could not reach %s - check your ' 'network connection' % host, error=name) # ------------------------------------------------------------ HTTP def _connection(self): cfg = self.cfg if cfg.scheme == 'https': if cfg.proxy is not None: conn = http.client.HTTPSConnection(cfg.proxy.host, cfg.proxy.port, timeout=cfg.timeout, context=self._ssl) conn.set_tunnel(cfg.host, cfg.port, headers=dict(cfg.proxy.headers)) return conn return http.client.HTTPSConnection(cfg.host, cfg.port, timeout=cfg.timeout, context=self._ssl) return http.client.HTTPConnection(cfg.host, cfg.port, timeout=cfg.timeout) def _open(self, call, method, headers, body): """Send one HTTP request (never following a redirect: http.client does not, so a 3xx comes back as a status and the Authorization header can never be replayed to another host). Returns (connection, response).""" conn = self._connection() if call is not None and not call.attach(conn): raise Cancelled() try: conn.connect() conn._seecrime_sock = conn.sock if call is not None and call.cancelled: raise Cancelled() conn.request(method, self.cfg.target, body=body, headers=headers) resp = conn.getresponse() except Cancelled: call.detach(conn) conn.close() raise except Exception as exc: if call is not None: call.detach(conn) conn.close() if call is not None and call.cancelled: raise Cancelled() raise self._network_error(exc) return conn, resp def _read_body(self, call, resp, cap, truncate): declared = resp.getheader('Content-Length') if not truncate and declared is not None and declared.strip().isdigit(): if int(declared) > cap: raise self._too_large() chunks = [] total = 0 try: while True: chunk = resp.read(min(65536, cap + 1 - total)) if not chunk: break chunks.append(chunk) total += len(chunk) if total > cap: if truncate: break raise self._too_large() except BridgeError: raise except Exception as exc: if call is not None and call.cancelled: raise Cancelled() if truncate: return b''.join(chunks) raise self._network_error(exc) return b''.join(chunks)[:cap] if truncate else b''.join(chunks) @staticmethod def _too_large(): return BridgeError(ERR_TOO_LARGE, 'seecrime MCP: the server\'s reply is larger than ' 'the bridge accepts (%d MB)' % (MAX_RESPONSE_BYTES // (1024 * 1024))) def _base_headers(self, accept='application/json, text/event-stream'): return { 'Authorization': 'Bearer ' + self.cfg.api_key, 'Accept': accept, 'User-Agent': self._user_agent, } def _headers_for(self, msg, kind): """Headers for a POST (Streamable HTTP, both eras). - Accept lists application/json and text/event-stream; Content-Type JSON. - MCP-Protocol-Version: a 2026-07-28 message's own _meta version (the header MUST match it); otherwise the version negotiated by initialize (2025-06-18+: on every request after initialization); none on initialize itself. - Mcp-Session-Id: the session a legacy server issued, if any; never on initialize (a new initialize starts a new session). - Mcp-Method on every 2026-07-28 request; Mcp-Name on tools/call, prompts/get (params.name) and resources/read (params.uri); and the Mcp-Param-* headers a tool's x-mcp-header annotations designate. Returns (headers, the session id sent or None). """ headers = self._base_headers() headers['Content-Type'] = 'application/json' method = msg.get('method') modern_version = _meta_protocol_version(msg) with self._state_lock: negotiated = self.negotiated_version session = self.session_id if method == 'initialize': version, session = None, None else: version = modern_version if modern_version is not None else negotiated if version is not None and _header_safe(version): headers['MCP-Protocol-Version'] = version if session: headers['Mcp-Session-Id'] = session if modern_version is not None and kind == 'request': if _header_safe(method): headers['Mcp-Method'] = method params = msg.get('params') if isinstance(msg.get('params'), dict) else {} name = None if method in ('tools/call', 'prompts/get'): name = params.get('name') elif method == 'resources/read': name = params.get('uri') if isinstance(name, str): headers['Mcp-Name'] = encode_header_value(name) if method == 'tools/call' and isinstance(name, str): for header, value in self._param_headers(name, params.get('arguments')): headers[header] = value return headers, session def _param_headers(self, tool, arguments): with self._tools_lock: annotations = self._tool_headers.get(tool) if not annotations or not isinstance(arguments, dict): return [] out = [] for path, header in annotations: found, value = _value_at(arguments, path) if not found or value is None: continue text = _param_text(value) if text is not None: out.append(('Mcp-Param-' + header, encode_header_value(text))) return out # ------------------------------------------------- one-way messages def _forward_oneway(self, msg, raw, retry_session=True): """POST a notification or a client response. Nothing is ever written to stdout for it: a 202 is the expected answer and a failure goes to stderr.""" label = _safe_label(msg.get('method') or 'a response') headers, session_used = self._headers_for(msg, 'oneway') try: conn, resp = self._open(None, 'POST', headers, raw) except BridgeError as exc: self.log.error('could not deliver %s: %s' % (label, exc.message)) return False try: status = resp.status ctype = _media_type(resp.getheader('Content-Type')) body = b'' if not 200 <= status < 300 and ctype != 'text/event-stream': body = self._read_body(None, resp, MAX_ERROR_BODY_BYTES, truncate=True) except BridgeError as exc: self.log.error('could not deliver %s: %s' % (label, exc.message)) return False finally: _close(None, conn, resp) if 200 <= status < 300: self.log.debug('%s -> HTTP %d' % (label, status)) return True if status == 404 and session_used and retry_session: if self._reestablish_session(session_used, None): return self._forward_oneway(msg, raw, retry_session=False) detail = self._excerpt(ctype, body) self.log.error('the server rejected %s (HTTP %d)%s' % (label, status, (': ' + detail) if detail else '')) return False def _handle_cancel(self, msg, raw): params = msg.get('params') if isinstance(msg.get('params'), dict) else {} request_id = params.get('requestId') with self._calls_lock: call = self._calls.get(_id_key(request_id)) if call is None: # Unknown, or already answered. A legacy server gets it anyway # (it is the legacy cancellation signal, and harmless if late). if _meta_protocol_version(msg) is None and self.negotiated_version is not None: self._forward_oneway(msg, raw) return started = call.cancel() reason = params.get('reason') self.log.info('request %s cancelled by the client%s' % ( _safe_label(request_id), (': ' + _safe_label(reason)) if isinstance(reason, str) else '')) if not call.modern and started: # 2025-11-25: disconnecting is NOT cancellation; the notification is. self._forward_oneway(msg, raw) # 2026-07-28: closing the response stream IS the cancellation. call.abort() def _cancel_upstream(self, call): """A legacy request the bridge stopped waiting for: tell the server, as the 2025-11-25 lifecycle asks of a sender whose request timed out. (A 2026-07-28 request needs nothing more: its stream is already closed.)""" note = {'jsonrpc': '2.0', 'method': 'notifications/cancelled', 'params': {'requestId': call.id, 'reason': 'timed out in the stdio bridge'}} self._forward_oneway(note, dumps_line(note).encode('ascii'), retry_session=False) # --------------------------------------------- legacy session state def _capture_initialize(self, call, result, resp): version = result.get('protocolVersion') session = resp.getheader('Mcp-Session-Id') if resp is not None else None if session is not None: session = session.strip() if not _header_safe(session) or len(session) > 8192: self.log.warn('ignored a session id from the server: it is not visible ASCII') session = None with self._state_lock: if isinstance(version, str) and _header_safe(version): self.negotiated_version = version self.session_id = session if not call.internal: params = call.msg.get('params') self._init_params = params if isinstance(params, dict) else None self.log.debug('initialized: protocol %s%s' % ( _safe_label(version), ', with a session' if session else '')) def _internal_id(self): with self._state_lock: self._internal_ids += 1 return 'seecrime-bridge-%d' % self._internal_ids def _run_internal(self, call): """POST a request the bridge makes itself; its replies are collected, never written to stdout. Returns the final response or None.""" headers, _ = self._headers_for(call.msg, 'request') outcome = self._post(call, headers) if outcome is not None: return None for msg in call.collected: if self._is_final(call, msg): return msg return None def _reestablish_session(self, session_used, parent): """2025-11-25: a 404 to a request carrying a session id means the session ended, and the client MUST start a new one with a fresh initialize. The stdio client cannot, so the bridge does, replaying the client's own initialize parameters, then sends notifications/initialized.""" with self._reinit_lock: with self._state_lock: if self.session_id != session_used: return True # another request already started a new session params = self._init_params self.session_id = None if params is None: return False self.log.info('the server ended the MCP session (HTTP 404); starting a new one') init = {'jsonrpc': '2.0', 'id': self._internal_id(), 'method': 'initialize', 'params': params} call = Call(init, dumps_line(init).encode('ascii'), internal=True, parent=parent) try: final = self._run_internal(call) except (BridgeError, Cancelled) as exc: self.log.error('could not start a new session (%s)' % type(exc).__name__) return False if not final or 'result' not in final: self.log.error('could not start a new session: the server refused initialize') return False note = {'jsonrpc': '2.0', 'method': 'notifications/initialized'} self._forward_oneway(note, dumps_line(note).encode('ascii'), retry_session=False) return True # ------------------------------------------- 2026 x-mcp-header support def _filter_tools(self, msg): """Record each tool's x-mcp-header annotations, and drop from the list a tool whose annotations are invalid -- a client using Streamable HTTP MUST exclude it from tools/list (2026-07-28, "Schema Extension").""" result = msg['result'] tools = result.get('tools') if not isinstance(tools, list): return msg kept = [] for tool in tools: if not isinstance(tool, dict): kept.append(tool) continue name = tool.get('name') annotations, problem = header_annotations(tool.get('inputSchema')) with self._tools_lock: if isinstance(name, str): if problem is None and annotations: self._tool_headers[name] = annotations else: self._tool_headers.pop(name, None) if problem is not None: self.log.warn('excluded tool %s from tools/list: %s' % (_safe_label(name), problem)) continue kept.append(tool) if len(kept) == len(tools): return msg new_result = dict(result) new_result['tools'] = kept new_msg = dict(msg) new_msg['result'] = new_result return new_msg def _refresh_tools(self, parent): """Re-read tools/list (every page) with the call's own _meta.""" params = parent.msg.get('params') if isinstance(parent.msg.get('params'), dict) else {} meta = params.get('_meta') if isinstance(params.get('_meta'), dict) else {} meta = {k: v for k, v in meta.items() if k.startswith('io.modelcontextprotocol/')} cursor = None for _ in range(50): list_params = {'_meta': meta} if cursor is not None: list_params['cursor'] = cursor req = {'jsonrpc': '2.0', 'id': self._internal_id(), 'method': 'tools/list', 'params': list_params} call = Call(req, dumps_line(req).encode('ascii'), internal=True, parent=parent) try: final = self._run_internal(call) except BridgeError: return False if not final or not isinstance(final.get('result'), dict): return False cursor = final['result'].get('nextCursor') if not isinstance(cursor, str) or not cursor: return True return True # --------------------------------------------------------- shutdown def _shutdown(self, drain): """stdin closed (the stdio spec's shutdown signal) or stdout went away. On EOF the requests already received are allowed to finish -- for up to SEECRIME_MCP_TIMEOUT -- so `echo '{...}' | bridge` still gets its answer; long-lived subscription streams are closed at once. Then the legacy session, if any, is ended with DELETE. """ with self._calls_lock: calls = list(self._calls.values()) for call in calls: if not drain or call.method == 'subscriptions/listen': call.cancel() call.abort() pending = self._futures.copy() if pending: concurrent.futures.wait(pending, timeout=self.cfg.timeout if drain else 1.0) with self._calls_lock: calls = list(self._calls.values()) for call in calls: call.cancel() call.abort() pending = self._futures.copy() if pending: concurrent.futures.wait(pending, timeout=2.0) self._pool.shutdown(wait=False) with self._state_lock: session = self.session_id version = self.negotiated_version if session: headers = self._base_headers() headers['Mcp-Session-Id'] = session if version: headers['MCP-Protocol-Version'] = version try: conn = self._connection() conn.timeout = min(5.0, self.cfg.timeout) conn.request('DELETE', self.cfg.target, headers=headers) resp = conn.getresponse() status = resp.status _close(None, conn, resp) self.log.debug('ended the MCP session (HTTP %d)' % status) except Exception as exc: self.log.debug('could not end the MCP session (%s)' % type(exc).__name__) def _debug_traceback(self): if self.log.verbose: for line in traceback.format_exc().splitlines(): self.log.debug(line) # ---------------------------------------------------------------------- CLI def claude_desktop_snippet(): entry = { 'mcpServers': { 'seecrime': { 'command': sys.executable or 'python', 'args': [os.path.abspath(__file__)], 'env': {'SEECRIME_API_KEY': 'sck_your_key_here'}, } } } return json.dumps(entry, indent=2) def mask_cli_message(message): """An argparse error with every argument VALUE masked and any API key redacted. argparse echoes what it could not parse, and stderr lands in the MCP client's log, which users paste into support requests: a mistyped `--key sck_...` would otherwise print the key there.""" message = re.sub(r'(--?[A-Za-z0-9][\w-]*)=\S+', r'\1=', message) match = re.match(r'(?s)^(unrecognized arguments: )(.*)$', message) if match: shown = [] for token in match.group(2).split(' '): if token.startswith('-'): shown.append(token) elif token: shown.append('') message = match.group(1) + ' '.join(shown) return _ANY_KEY_RE.sub('[REDACTED]', message) class _Parser(argparse.ArgumentParser): def error(self, message): self.print_usage(sys.stderr) self.exit(2, '%s: error: %s\n' % (self.prog, mask_cli_message(message))) def build_parser(): epilog = '\n'.join([ 'configuration (a command-line option overrides the environment variable):', ' SEECRIME_API_KEY your seecrime API key (required); create one at', ' https://www.seecrime.com/developers', ' SEECRIME_MCP_URL the MCP endpoint (default %s)' % DEFAULT_URL, ' SEECRIME_MCP_TIMEOUT seconds to wait for the server (default %g)' % DEFAULT_TIMEOUT, '', 'Claude Desktop: add this to claude_desktop_config.json and restart it:', '', claude_desktop_snippet(), '', 'Put the key in "env", not in "args": a command line is visible to other', 'programs on this computer. The key is only ever sent over TLS, in the', 'Authorization header. Never commit it.', ]) parser = _Parser( description='stdio bridge to the seecrime.com MCP server: reads JSON-RPC ' 'messages on stdin, forwards them over HTTPS, and writes the ' 'replies to stdout.', epilog=epilog, formatter_class=argparse.RawDescriptionHelpFormatter) parser.add_argument('--api-key', metavar='KEY', help='API key (overrides SEECRIME_API_KEY; prefer the variable)') parser.add_argument('--url', metavar='URL', help='MCP endpoint (overrides SEECRIME_MCP_URL)') parser.add_argument('--timeout', metavar='SECONDS', help='seconds to wait for the server (overrides SEECRIME_MCP_TIMEOUT)') parser.add_argument('--verbose', action='store_true', help='log each request and its outcome to stderr') parser.add_argument('--version', action='version', version='seecrime-mcp-stdio %s' % __version__) return parser def _say(text): """One line to stderr, before the Log exists (never the key).""" try: sys.stderr.write('seecrime-mcp: %s\n' % text) sys.stderr.flush() except Exception: pass def main(argv=None, environ=None): parser = build_parser() args = parser.parse_args(argv) environ = os.environ if environ is None else environ try: cfg = load_config(args, environ) except ConfigError as exc: _say(str(exc)) return 2 if sys.stdin is None or sys.stdout is None: _say('stdin and stdout must be connected to the MCP client') return 2 redactor = Redactor([cfg.api_key]) try: cfg.proxy = find_proxy(cfg.scheme, cfg.host, Log(redactor)) except ConfigError as exc: _say(str(exc)) return 2 if not cfg.api_key.startswith('sck_'): _say('warning: SEECRIME_API_KEY does not start with "sck_", so it is probably ' 'not a seecrime API key') bridge = Bridge(cfg, stdin=sys.stdin.buffer, stdout=sys.stdout.buffer) return bridge.run() def _exit(code): for stream in (sys.stdout, sys.stderr): try: stream.flush() except Exception: pass os._exit(code) if __name__ == '__main__': try: _code = main() except SystemExit: raise # --help / --version / an argparse usage error except KeyboardInterrupt: _code = 130 except BaseException as _exc: # never a traceback: it could carry a header value _say('fatal error (%s)' % type(_exc).__name__) _code = 1 _exit(_code)