flash_mcp.py
OT Flash r56 · Online reference / 在线参考
#!/usr/bin/env python3
"""OT Flash r56 minimal MCP client. Standard library only, no automatic writes.
Environment: FLASH_BASE_URL, FLASH_ACCESS_TOKEN.
Do not put tokens on the command line or commit them to examples.
"""
import argparse,base64,json,os,re,sys,urllib.error,urllib.parse,urllib.request,uuid
VERSIONS=('2026-07-28','2025-11-25','2025-06-18','2025-03-26')
class ClientError(Exception):pass
class NoRedirects(urllib.request.HTTPRedirectHandler):
def redirect_request(self,request,fp,code,msg,headers,newurl):
raise ClientError('Redirect rejected; verify the exact trusted MCP URL before sending a bearer token.')
def envelope(method,params=None,protocol='2026-07-28',request_id=None):
params=dict(params or {})
if protocol=='2026-07-28':
params['_meta']={'io.modelcontextprotocol/protocolVersion':protocol,
'io.modelcontextprotocol/clientInfo':{'name':'ot-flash-manual-client','version':'1.0.0'},
'io.modelcontextprotocol/clientCapabilities':{}}
body={'jsonrpc':'2.0','id':request_id or uuid.uuid4().hex,'method':method,'params':params}
headers={'Content-Type':'application/json','Accept':'application/json, text/event-stream','MCP-Protocol-Version':protocol}
if protocol=='2026-07-28':
headers['Mcp-Method']=method
name=params.get('name') if method=='tools/call' else params.get('uri') if method=='resources/read' else None
if name:
headers['Mcp-Name']=name if name.isascii() else '=?base64?'+base64.b64encode(name.encode()).decode()+'?='
return body,headers
def reject_placeholders(value):
if isinstance(value,str) and re.search(r'<[A-Z][A-Z0-9_ ]*>',value):raise ClientError('Replace every example placeholder with an actual authorized server value before calling.')
if isinstance(value,dict):
for child in value.values():reject_placeholders(child)
if isinstance(value,list):
for child in value:reject_placeholders(child)
class Client:
def __init__(self,base,token,endpoint,protocol,allow_http=False):
parsed=urllib.parse.urlsplit(base)
if parsed.scheme not in ('https','http') or not parsed.hostname or parsed.username or parsed.password or parsed.query or parsed.fragment:
raise ClientError('Provide an exact trusted HTTP(S) base URL without credentials, query or fragment.')
if parsed.scheme!='https' and not allow_http:raise ClientError('HTTPS required; --allow-http is only for an explicitly approved isolated test site.')
if not token or any(c in token for c in '\r\n'):raise ClientError('Set FLASH_ACCESS_TOKEN to a valid access token.')
self.url=base.rstrip('/')+('/mcp/business' if endpoint=='business' else '/mcp');self.token=token;self.protocol=protocol
self.http=urllib.request.build_opener(urllib.request.ProxyHandler({}),NoRedirects());self.initialized=False
def rpc(self,method,params=None):
body,headers=envelope(method,params,self.protocol);headers['Authorization']='Bearer '+self.token
data=json.dumps(body,ensure_ascii=False,separators=(',',':')).encode();request=urllib.request.Request(self.url,data=data,headers=headers,method='POST')
try:
with self.http.open(request,timeout=90) as response:
if response.status==202:return {'accepted':True,'note':'Notification accepted; no business completion is implied.'}
raw=response.read(32*1024*1024+1)
if len(raw)>32*1024*1024:raise ClientError('Response limit exceeded; use bounded queries or pages.')
if 'application/json' not in response.headers.get('Content-Type',''):raise ClientError('Expected JSON; this r56 endpoint does not require an SSE parser.')
value=json.loads(raw)
except urllib.error.HTTPError as error:
raise ClientError('HTTP '+str(error.code)+'. Check OAuth, permissions, protocol headers and exact endpoint; no mutation was retried.') from None
except urllib.error.URLError:
raise ClientError('Network or TLS request failed. Read the original status before retrying any mutation.') from None
if value.get('id')!=body['id']:raise ClientError('JSON-RPC response id mismatch.')
if 'error' in value:raise ClientError('JSON-RPC '+str(value['error'].get('code'))+': '+str(value['error'].get('message','Error')))
result=value.get('result')
if not isinstance(result,dict):raise ClientError('Missing JSON-RPC result object.')
if result.get('isError'):
# Return the original structured failure for the caller to inspect;
# never retry automatically, especially for uncertain writes.
print(json.dumps(result,ensure_ascii=False,indent=2));raise ClientError('Tool reported isError=true; inspect recovery fields above.')
return result
def initialize(self):
if not self.initialized and self.protocol!='2026-07-28':
self.rpc('initialize',{'protocolVersion':self.protocol,'capabilities':{},'clientInfo':{'name':'ot-flash-manual-client','version':'1.0.0'}})
self.initialized=True
def self_test():
modern,headers=envelope('tools/call',{'name':'dfc_ontologies','arguments':{'page':1,'pageSize':20}},request_id='offline-check')
assert headers['Mcp-Method']==modern['method'] and headers['Mcp-Name']==modern['params']['name']
assert headers['MCP-Protocol-Version']==modern['params']['_meta']['io.modelcontextprotocol/protocolVersion']
legacy,legacy_headers=envelope('tools/list',{},'2025-11-25',request_id='legacy-check')
assert '_meta' not in legacy['params'] and 'Mcp-Method' not in legacy_headers
try:reject_placeholders({'id':'<ONTOLOGY_ID>'})
except ClientError:pass
else:raise AssertionError('placeholder guard failed')
print('PASS: offline request construction and placeholder guard. No network or product acceptance was executed.')
def main():
parser=argparse.ArgumentParser(description=__doc__)
parser.add_argument('--self-test',action='store_true');parser.add_argument('--endpoint',choices=['ontology','business'],default='ontology')
parser.add_argument('--protocol',choices=VERSIONS,default=VERSIONS[0]);parser.add_argument('--allow-http',action='store_true')
parser.add_argument('command',nargs='?',choices=['discover','tools','resources','read-resource','call']);parser.add_argument('name',nargs='?')
group=parser.add_mutually_exclusive_group();group.add_argument('--arguments');group.add_argument('--arguments-file')
args=parser.parse_args()
if args.self_test:return self_test()
if not args.command:parser.error('Choose a command or --self-test.')
client=Client(os.environ.get('FLASH_BASE_URL',''),os.environ.get('FLASH_ACCESS_TOKEN',''),args.endpoint,args.protocol,args.allow_http)
if args.command=='discover':
result=client.rpc('server/discover',{}) if args.protocol=='2026-07-28' else client.rpc('initialize',{'protocolVersion':args.protocol,'capabilities':{},'clientInfo':{'name':'ot-flash-manual-client','version':'1.0.0'}})
else:
client.initialize()
if args.command=='tools':result=client.rpc('tools/list',{})
elif args.command=='resources':result=client.rpc('resources/list',{})
elif args.command=='read-resource':
if not args.name:parser.error('read-resource needs an actual resource URI.')
result=client.rpc('resources/read',{'uri':args.name})
else:
if not args.name:parser.error('call needs an actual tool name.')
raw=args.arguments or '{}'
if args.arguments_file:
with open(args.arguments_file,encoding='utf-8') as stream:raw=stream.read(4*1024*1024+1)
if len(raw)>4*1024*1024:raise ClientError('Arguments file limit exceeded.')
value=json.loads(raw)
if isinstance(value,dict) and 'arguments' in value:
if value.get('name')!=args.name:raise ClientError('Request file tool name does not match the command.')
value=value['arguments']
if not isinstance(value,dict):raise ClientError('Tool arguments must be a JSON object.')
reject_placeholders(value);result=client.rpc('tools/call',{'name':args.name,'arguments':value})
print(json.dumps(result,ensure_ascii=False,indent=2))
if __name__=='__main__':
try:main()
except (ClientError,ValueError,OSError) as error:
print(str(error),file=sys.stderr);raise SystemExit(2)