#!/usr/bin/env python3 """Dependency-free, newline-delimited stdio MCP bridge for Slivervow Gomoku.""" import argparse import fcntl import json import os from pathlib import Path import re import sys import tempfile import threading import time import urllib.error import urllib.parse import urllib.request DEFAULT_URL = 'https://slivervow.online' PROTOCOLS = ('2025-11-25', '2025-06-18', '2025-03-26', '2024-11-05') WAIT_SEGMENT = 25 INSTRUCTIONS = ('Only join after a user invitation. You are white (2); board rows use .=empty X=black O=white, coordinates row/col 1..15. ' 'Join/read_state without after_seq gives the initial full board. On your turn read_state(after_seq=latest seq,claim_turn=true) when you retain board context, then act once with latest expected_version and a unique action_id. ' 'Retry uncertain actions only with identical action_id and payload. On conflicts sync before choosing a new action. ' 'After acting use wait_events(after_seq=returned seq, timeout_seconds=300); apply events to your known board and check turn/status. ' 'After context loss or resync_required explicitly read_state without a cursor; never guess missing board context. ' 'Pending rematch_by pauses placement. A user rematch request from the webpage authorizes accept_rematch; never reset an active board without a request. Use rematch to request during play or immediately restart after terminal only on user request. decline_rematch is opponent-only; cancel_rematch requester-only. On restarted discard the old board and moves, initialize an empty board, then apply subsequent events; round increases and seq/version never reset. ' 'Stop on won/draw/resigned or user cancellation. A 300-second timeout may suspend the session; no automatic resignation or automatic wakeup. ' 'Do not busy-loop. list_rooms recovers locally cached IDs; another machine needs an invitation. Resign only on explicit user request.') def schema(properties, required): return {'type': 'object', 'properties': properties, 'required': required, 'additionalProperties': False} GAME = {'type': 'string', 'pattern': '^[A-Za-z0-9_-]{1,128}$'} VERSION = {'type': 'integer', 'minimum': 0} COORD = {'type': 'integer', 'minimum': 1, 'maximum': 15} TOOLS = [ {'name': 'join_room', 'description': 'Join a user-invited Gomoku room as white; repeated invitations recover the same seat. Returns initial compact board; follow the server turn workflow.', 'inputSchema': schema({'invite_code': {'type': 'string', 'minLength': 1, 'maxLength': 256}}, ['invite_code'])}, {'name': 'get_game', 'description': 'Read current board, own color, turn, status and version. Use after a stale-version error.', 'inputSchema': schema({'game_id': GAME}, ['game_id'])}, {'name': 'play_move', 'description': 'Place your stone at row/col 1..15 using the exact latest version. Only play your turn; never blindly retry after a network error: read first.', 'inputSchema': schema({'game_id': GAME, 'row': COORD, 'col': COORD, 'version': VERSION}, ['game_id', 'row', 'col', 'version'])}, {'name': 'wait_for_turn', 'description': 'Wait for any version change, at most 25 seconds. Check turn and status on return. Suspend on timeout or cancellation; no automatic session wakeup.', 'inputSchema': schema({'game_id': GAME, 'after_version': VERSION, 'timeout_seconds': {'type': 'integer', 'minimum': 1, 'maximum': 25, 'default': 25}}, ['game_id', 'after_version'])}, {'name': 'resign_game', 'description': 'Concede the game only when the user asks to resign; use the latest version. A request to stop waiting is not a request to resign.', 'inputSchema': schema({'game_id': GAME, 'version': VERSION}, ['game_id', 'version'])}, ] TOOLS += [ {'name':'read_state','description':'Omit after_seq only for initial state, context loss or resync_required. Otherwise pass latest seq for deltas. Set claim_turn=true on your turn to acknowledge for 120 seconds.', 'inputSchema':schema({'game_id':GAME,'after_seq':VERSION,'claim_turn':{'type':'boolean','default':False}},['game_id'])}, {'name':'act','description':'One action with version protection and idempotent action_id. Repeat identical ID/payload after uncertain network failure. Resign/rematch only on explicit user request; a webpage rematch request authorizes accept_rematch. Pending requests pause placement. restarted requires clearing the old board.', 'inputSchema':schema({'game_id':GAME,'expected_version':VERSION,'action_id':{'type':'string','minLength':1,'maxLength':128},'action':{'type':'object','oneOf':[schema({'type':{'type':'string','const':'place'},'row':COORD,'col':COORD},['type','row','col']),*[schema({'type':{'type':'string','const':kind}},['type']) for kind in ('resign','rematch','accept_rematch','decline_rematch','cancel_rematch')]]}},['game_id','expected_version','action_id','action'])}, {'name':'wait_events','description':'Wait up to 300 seconds for durable events; internal HTTP segments <=25 seconds. timeout_seconds=0 polls once. No board on timeout. Stop on cancellation/terminal; timeout may suspend, never resign or promise automatic wakeup.', 'inputSchema':schema({'game_id':GAME,'after_seq':VERSION,'timeout_seconds':{'type':'integer','minimum':0,'maximum':300,'default':300}},['game_id','after_seq'])}, {'name':'list_rooms','description':'List locally cached room IDs without credentials. Explicitly read_state to recover board context; another machine requires invitation.', 'inputSchema':schema({},[])} ] class ToolError(Exception): pass def validate(name, args): tool = next((t for t in TOOLS if t['name'] == name), None) if not tool: raise ToolError('Unknown tool.') s = tool['inputSchema'] if not isinstance(args, dict) or set(args) - set(s['properties']) or not set(s['required']) <= set(args): raise ToolError('Invalid tool arguments: missing or unexpected fields.') for key, value in args.items(): rule = s['properties'][key] if rule['type'] == 'integer': if type(value) is not int or value < rule.get('minimum', 0) or value > rule.get('maximum', 2**53 - 1): raise ToolError('Invalid integer argument: ' + key) elif rule['type'] == 'boolean': if type(value) is not bool: raise ToolError('Invalid boolean argument: '+key) elif rule['type'] == 'object': if not isinstance(value,dict) or value.get('type') not in ('place','resign','rematch','accept_rematch','decline_rematch','cancel_rematch') or set(value)!=({'type','row','col'} if value.get('type')=='place' else {'type'}): raise ToolError('Invalid action.') if value['type']=='place' and any(type(value[k]) is not int or not 1<=value[k]<=15 for k in ('row','col')): raise ToolError('Invalid action coordinates.') elif not isinstance(value, str) or not value or len(value) > rule.get('maxLength', 128) or ('pattern' in rule and not re.fullmatch(rule['pattern'], value)): raise ToolError('Invalid string argument: ' + key) class Cache: def __init__(self, path): self.path = Path(path).expanduser() self.lock = threading.Lock() def access(self, game_id=None, token=None): with self.lock: self.path.parent.mkdir(parents=True, exist_ok=True, mode=0o700) # Never follow a symlink to a credential or lock file. fd = os.open(str(self.path) + '.lock', os.O_CREAT | os.O_RDWR | os.O_NOFOLLOW, 0o600) with os.fdopen(fd, 'r+') as lockfile: os.fchmod(lockfile.fileno(), 0o600) fcntl.flock(lockfile, fcntl.LOCK_EX) try: fd = os.open(self.path, os.O_RDONLY | os.O_NOFOLLOW) with os.fdopen(fd) as source: os.fchmod(source.fileno(), 0o600) rooms = json.load(source) except FileNotFoundError: rooms = {} if not isinstance(rooms, dict): raise ToolError('Credential cache is invalid; restore it or use a fresh cache path.') if token is not None: rooms[game_id] = token tmpfd, tmp = tempfile.mkstemp(dir=self.path.parent, prefix='.rooms-') try: with os.fdopen(tmpfd, 'w') as output: json.dump(rooms, output) output.flush() os.fsync(output.fileno()) os.replace(tmp, self.path) finally: if os.path.exists(tmp): os.unlink(tmp) if game_id is None: return sorted(k for k in rooms if isinstance(k,str) and re.fullmatch(GAME['pattern'],k)) value = rooms.get(game_id) if not isinstance(value, str) or not value: raise ToolError('Room credentials not found. Join using the user invitation first.') return value class NoRedirect(urllib.request.HTTPRedirectHandler): def redirect_request(self, req, fp, code, msg, headers, newurl): return None def public_game(game): """Allowlist and validate all backend data before showing it to the model.""" try: gid = game['game_id'] if not isinstance(gid, str) or not re.fullmatch(GAME['pattern'], gid): raise ValueError() board = game['board'] if len(board) != 15 or any(len(row) != 15 or any(type(v) is not int or v not in (0, 1, 2) for v in row) for row in board): raise ValueError() if type(game['version']) is not int or game['version'] < 0 or game['status'] not in ('waiting', 'playing', 'won', 'draw', 'resigned'): raise ValueError() if any(type(game[k]) is not int or game[k] not in allowed for k, allowed in [('you', (1, 2)), ('turn', (1, 2)), ('winner', (0, 1, 2))]): raise ValueError() result = {k: game[k] for k in ('game_id', 'version', 'status', 'turn', 'winner', 'you')} for key, default, allowed in [('round',1,None),('rematch_by',0,(0,1,2))]: value=game.get(key,default) if type(value) is not int or (allowed is None and value<1) or (allowed is not None and value not in allowed): raise ValueError() result[key]=value result['seq'] = game.get('seq',game['version']) result['board'] = [''.join('.XO'[v] for v in row) for row in board] return result except (KeyError, TypeError, ValueError): raise ToolError('Invalid game response from service.') from None def public_v2(data): """Only copy known protocol fields; never echo arbitrary service strings.""" try: result = {} if not isinstance(data,dict) or not isinstance(data.get('game_id'),str) or not re.fullmatch(GAME['pattern'],data['game_id']): raise ValueError() result['game_id']=data['game_id'] for key in ('seq','version','turn','winner','you','round','rematch_by'): if key in data: if type(data[key]) is not int or not 0<=data[key]<=2**53-1: raise ValueError() if key=='round' and data[key]<1 or key=='rematch_by' and data[key] not in (0,1,2): raise ValueError() result[key]=data[key] if 'seq' not in result: raise ValueError() if 'status' in data: if data['status'] not in ('waiting','playing','won','draw','resigned'): raise ValueError() result['status']=data['status'] for flag in ('timeout','cancelled','resync_required'): if data.get(flag) is True: result[flag]=True if 'board' in data: if not isinstance(data['board'],list) or len(data['board'])!=15 or any(not isinstance(row,str) or not re.fullmatch(r'[.XO]{15}',row) for row in data['board']): raise ValueError() result['board']=data['board'] if 'events' in data: if not isinstance(data['events'],list) or len(data['events'])>256: raise ValueError() result['events']=[] for event in data['events']: if not isinstance(event,dict) or event.get('type') not in ('joined','place','resign','rematch_requested','rematch_declined','rematch_cancelled','restarted'): raise ValueError() clean={'type':event['type']} for key in ('seq','version','player','row','col','round'): if key in event: if type(event[key]) is not int or not 0<=event[key]<=2**53-1: raise ValueError() clean[key]=event[key] if clean['type']=='restarted' and ('round' not in clean or clean['round']<1): raise ValueError() result['events'].append(clean) return result except (TypeError,ValueError,KeyError): raise ToolError('Invalid game response from service.') from None class Connector: def __init__(self, base_url, cache_path): parsed = urllib.parse.urlsplit(base_url) if parsed.scheme not in ('http', 'https') or not parsed.netloc or parsed.username or parsed.password or parsed.query or parsed.fragment or parsed.path not in ('', '/'): raise ValueError('Service URL must be an HTTP(S) origin.') self.base_url = base_url.rstrip('/') self.cache = Cache(cache_path) self.opener = urllib.request.build_opener(NoRedirect()) def request(self, path, data=None, token=None, timeout=15): headers = {'Content-Type': 'application/json', 'Accept': 'application/json'} if token: headers['Authorization'] = 'Bearer ' + token req = urllib.request.Request(self.base_url + path, data=None if data is None else json.dumps(data).encode(), headers=headers) try: with self.opener.open(req, timeout=timeout) as response: raw = response.read(262145) if len(raw) > 262144: raise ToolError('Service response is too large.') return json.loads(raw) except urllib.error.HTTPError as exc: status = exc.code exc.close() hint = {400: 'Invalid game action.', 401: 'Room credentials expired or invalid.', 403: 'This action is not allowed for your seat.', 404: 'Room not found or expired.', 409: 'State conflict: sync with read_state before choosing a new action.', 410: 'Room expired.', 429: 'Rate limited: stop and try later.'}.get(status, 'Service request failed.') raise ToolError(f'HTTP {status}: {hint}') from None except (OSError, ValueError): raise ToolError('Network or response error. Retry act only with identical action_id and payload; otherwise sync with read_state.') from None def call(self, name, args, cancelled=None): validate(name, args) if name == 'list_rooms': return {'rooms':self.cache.access()} if name == 'join_room': data = self.request('/api/games/join', {'invite_code': args['invite_code']}) game = public_game(data.get('game')) token = data.get('token') if not isinstance(token, str) or not token or len(token) > 4096: raise ToolError('Service returned invalid room credentials.') self.cache.access(game['game_id'], token) return game gid = args['game_id'] token = self.cache.access(gid) path = '/api/games/' + gid if name == 'read_state': return public_v2(self.request(path+'/read-state',{k:v for k,v in args.items() if k!='game_id'},token)) if name == 'act': return public_v2(self.request(path+'/actions',{k:v for k,v in args.items() if k!='game_id'},token)) if name == 'wait_events': return self.wait_events(gid,token,args['after_seq'],args.get('timeout_seconds',300),cancelled) if name == 'play_move': data = self.request(path + '/moves', {k: args[k] for k in ('row', 'col', 'version')}, token) elif name == 'resign_game': data = self.request(path + '/resign', {'version': args['version']}, token) elif name == 'wait_for_turn': timeout = args.get('timeout_seconds', 25) data = self.request(path + '/wait?' + urllib.parse.urlencode({'after': args['after_version'], 'timeout': timeout}), token=token, timeout=timeout + 5) else: data = self.request(path, token=token) game = public_game(data.get('game')) if name in ('play_move','resign_game','wait_for_turn'): del game['board'] if name=='wait_for_turn' and not data.get('changed'): return dict(game_id=gid,seq=game['seq'],timeout=True,changed=False) # Legacy callers supply board version, not event cursor. Explicit resync # avoids pretending a partial legacy response is a complete delta. game['resync_required']=True return game def wait_events(self, gid, token, after, timeout, cancelled=None): deadline=time.monotonic()+timeout while True: if cancelled is not None and cancelled.is_set(): return dict(game_id=gid,seq=after,cancelled=True) segment=min(WAIT_SEGMENT,max(0,deadline-time.monotonic())) segment_started=time.monotonic() data=public_v2(self.request('/api/games/'+gid+'/events?'+urllib.parse.urlencode({'after_seq':after,'timeout':segment}),token=token,timeout=segment+5)) if cancelled is not None and cancelled.is_set(): return dict(game_id=gid,seq=after,cancelled=True) if not data.get('timeout') or time.monotonic()>=deadline: return data # A compliant backend waits for the whole segment. Defensive pacing # prevents a broken/early timeout response from spinning requests. delay=min(max(0,segment-(time.monotonic()-segment_started)),max(0,deadline-time.monotonic())) if cancelled is not None: cancelled.wait(delay) else: time.sleep(delay) def tool_result(data=None, error=None): return {'content': [{'type': 'text', 'text': error if error else json.dumps(data, ensure_ascii=False, separators=(',',':'))}], 'isError': error is not None} class Server: def __init__(self, connector, output=sys.stdout): self.connector, self.output = connector, output self.lock = threading.RLock() self.active = {} def send(self, rid, result=None, error=None): with self.lock: message = {'jsonrpc': '2.0', 'id': rid} message['error' if error else 'result'] = error if error else result self.output.write(json.dumps(message, ensure_ascii=False) + '\n') self.output.flush() def run_call(self, rid, params, cancelled): try: result = tool_result(error='Request cancelled.') if cancelled.is_set() else tool_result(self.connector.call(params['name'], params.get('arguments', {}), cancelled)) except ToolError as exc: result = tool_result(error=str(exc)) except Exception: # Never echo exception text that may include tokens, invites or response bodies. result = tool_result(error='Local connector error. Check cache permissions and service configuration.') with self.lock: try: if not cancelled.is_set(): self.send(rid, result=result) finally: self.active.pop(rid, None) def handle(self, msg): if not isinstance(msg, dict) or msg.get('jsonrpc') != '2.0' or not isinstance(msg.get('method'), str) or ('id' in msg and (type(msg['id']) not in (int, str) or msg['id'] is None)): self.send(None, error={'code': -32600, 'message': 'Invalid Request'}) return method, params, rid = msg['method'], msg.get('params', {}), msg.get('id') if not isinstance(params, dict): if rid is not None: self.send(rid, error={'code': -32602, 'message': 'Invalid params'}) return if method == 'notifications/cancelled': with self.lock: request_id = params.get('requestId') if type(request_id) in (str, int) and request_id in self.active: self.active[request_id].set() return if rid is None: return if method == 'initialize': version = params.get('protocolVersion') self.send(rid, result={'protocolVersion': version if version in PROTOCOLS else PROTOCOLS[0], 'capabilities': {'tools': {'listChanged': False}}, 'serverInfo': {'name': 'slivervow-games', 'version': '2.0.0'}, 'instructions': INSTRUCTIONS}) elif method == 'ping': self.send(rid, result={}) elif method == 'tools/list': self.send(rid, result={'tools': TOOLS}) elif method == 'tools/call': if not isinstance(params.get('name'), str): self.send(rid, error={'code': -32602, 'message': 'Tool name is required'}) return with self.lock: if rid in self.active or len(self.active) >= 4: self.send(rid, error={'code': -32000, 'message': 'Connector busy; at most four active requests'}) return cancelled = threading.Event() self.active[rid] = cancelled threading.Thread(target=self.run_call, args=(rid, params, cancelled), daemon=True).start() else: self.send(rid, error={'code': -32601, 'message': 'Method not found'}) def serve(self, stream=sys.stdin): for line in stream: try: msg = json.loads(line) except ValueError: self.send(None, error={'code': -32700, 'message': 'Parse error'}) continue self.handle(msg) with self.lock: for cancelled in self.active.values(): cancelled.set() def main(): parser = argparse.ArgumentParser(description=__doc__) parser.add_argument('--cache-file', default=os.environ.get('SLIVERVOW_GAMES_CACHE', '~/.local/share/slivervow-games/rooms.json')) commands=parser.add_subparsers(dest='command') wait=commands.add_parser('wait-events') wait.add_argument('--game',required=True) wait.add_argument('--after',required=True,type=int) wait.add_argument('--timeout',type=int,default=300) args = parser.parse_args() connector = Connector(os.environ.get('SLIVERVOW_GAMES_BASE_URL', DEFAULT_URL), args.cache_file) if args.command=='wait-events': try: result=connector.call('wait_events',dict(game_id=args.game,after_seq=args.after,timeout_seconds=args.timeout)) except KeyboardInterrupt: result=dict(game_id=args.game,seq=args.after,cancelled=True) except ToolError as exc: result={'error':str(exc)} except Exception: result={'error':'Local connector error. Check cache permissions and service configuration.'} print(json.dumps(result,ensure_ascii=False,separators=(',',':'))) if 'error' in result: raise SystemExit(1) else: Server(connector).serve() if __name__ == '__main__': main()