"""Dependency-free MCP stdio server, MCP 2025-03-26. No arbitrary SQL/file tools.""" import json import sys from pathlib import Path from engine import Engine, Rejected, volume_breaker from alpaca_reader import paper_account_check, crypto_quotes def tool(name, description, properties=None, required=None, read_only=True): return {'name':name, 'description':description, 'inputSchema':{'type':'object','properties':properties or {},'required':required or [],'additionalProperties':False}, 'annotations':{'readOnlyHint':read_only,'destructiveHint':False,'idempotentHint':read_only,'openWorldHint':name in ('paper_account_check','crypto_quotes')}} S = {'type':'string'} O = {'type':'object'} TOOLS = [ tool('quant_status','Inspect local simulation account, health and disabled execution modes.'), tool('initialize_simulation','Create a local virtual account once; no broker account or funds.', {'virtual_cash':S}, ['virtual_cash'],False), tool('evaluate_risk','Validate evidence structure and calculate local long-only sizing. May update session/drawdown state.', {'symbol':S,'entry':S,'stop':S,'quote_at':S,'reasoning':O}, ['symbol','entry','stop','quote_at','reasoning'],False), tool('simulate_entry','Record an atomic local simulated fill with research and idempotency key; no broker call.', {'request_id':S,'symbol':S,'entry':S,'stop':S,'quote_at':S,'reasoning':O}, ['request_id','symbol','entry','stop','quote_at','reasoning'],False), tool('mark_positions','Apply fresh complete quotes, simulate stop exits and latch drawdown circuit breakers.', {'quotes':O},['quotes'],False), tool('recover_session','Recover a latched daily halt only after a later UTC day with no open positions.',read_only=False), tool('performance_report','Closed local simulated lots and economic PnL after modeled fees; no tax adjustments.',{'days':{'type':'integer'}},['days']), tool('trial_review','Report unmet trial conditions; can never grant live clearance.'), tool('tax_workpaper','Return annual simulation disposition records, not an IRS form or import file.',{'year':{'type':'integer'}},['year']), tool('volume_safety','Check 20 preceding closed volume bars; undefined variance halts.',{'history':{'type':'array','items':S},'current':S},['history','current']), tool('paper_account_check','Authenticate read-only against exact Alpaca paper endpoint with environment credentials.'), tool('crypto_quotes','Read current BTC/USD and ETH/USD bid/ask quotes. No fallback data.'), ] def validate(schema, value): if not isinstance(value,dict) or set(value)-set(schema['properties']) or set(schema['required'])-set(value): raise Rejected('Invalid tool arguments') for key,val in value.items(): kind = schema['properties'][key]['type'] if not ((kind=='string' and isinstance(val,str)) or (kind=='object' and isinstance(val,dict)) or (kind=='array' and isinstance(val,list) and all(isinstance(x,str) for x in val)) or (kind=='integer' and isinstance(val,int) and not isinstance(val,bool))): raise Rejected('Invalid argument type: '+key) class Server: def __init__(self): self.engine = Engine() self.initialized = False self.ready = False def dispatch(self, method, params): if method=='initialize': if self.initialized: raise Rejected('Already initialized') if not isinstance(params, dict) or params.get('protocolVersion') not in ('2024-11-05','2025-03-26','2025-06-18','2025-11-25'): raise Rejected('Unsupported protocol') self.initialized=True return {'protocolVersion':'2025-03-26','capabilities':{'tools':{}}, 'serverInfo':{'name':'foresixa-quant','version':'0.1.0'}, 'instructions':Path(__file__).parent.parent.joinpath('cognitive_bot.md').read_text()} if method=='notifications/initialized': self.ready = self.initialized return None if method=='ping': return {} if not self.ready: raise Rejected('Initialize session first') if method=='tools/list': return {'tools':TOOLS} if method!='tools/call': raise LookupError('Method not found') if not isinstance(params,dict): raise Rejected('Parameters must be an object') name=params.get('name') spec=next((t for t in TOOLS if t['name']==name),None) if not spec: raise Rejected('Unknown tool') arguments=params.get('arguments',{}) try: validate(spec['inputSchema'],arguments) functions={'quant_status':self.engine.status,'initialize_simulation':self.engine.initialize, 'evaluate_risk':self.engine.evaluate,'simulate_entry':self.engine.simulate_entry, 'mark_positions':self.engine.mark,'recover_session':self.engine.recover, 'performance_report':self.engine.report,'trial_review':self.engine.audit, 'tax_workpaper':self.engine.tax_export,'volume_safety':volume_breaker, 'paper_account_check':paper_account_check,'crypto_quotes':crypto_quotes} result=functions[name](**arguments) return {'content':[{'type':'text','text':json.dumps(result)}],'isError':False} except (Rejected, KeyError, TypeError, ValueError) as exc: message=str(exc) if isinstance(exc,Rejected) else 'Invalid data; operation rejected' return {'content':[{'type':'text','text':message}],'isError':True} def handle(self, request): if not isinstance(request,dict) or request.get('jsonrpc')!='2.0' or not isinstance(request.get('method'),str): return {'jsonrpc':'2.0','id':None,'error':{'code':-32600,'message':'Invalid request'}} notification='id' not in request try: result=self.dispatch(request['method'],request.get('params',{})) reply={'jsonrpc':'2.0','id':request.get('id'),'result':result} except LookupError: reply={'jsonrpc':'2.0','id':request.get('id'),'error':{'code':-32601,'message':'Method not found'}} except Rejected as exc: reply={'jsonrpc':'2.0','id':request.get('id'),'error':{'code':-32602,'message':str(exc)}} except Exception: reply={'jsonrpc':'2.0','id':request.get('id'),'error':{'code':-32603,'message':'Internal failure; inspect local runtime'}} return None if notification else reply def main(): server=Server() for line in sys.stdin: try: if len(line)>1_000_000: raise ValueError('Too large') req=json.loads(line) if isinstance(req,list): replies=[server.handle(x) for x in req] reply=[x for x in replies if x is not None] or None else: reply=server.handle(req) except (ValueError,RecursionError): reply={'jsonrpc':'2.0','id':None,'error':{'code':-32700,'message':'Parse error'}} if reply is not None: print(json.dumps(reply),flush=True) if __name__=='__main__': main()