Geef protocolfouten en toolfouten hun eigen laag

import copy

import json

import math

import re

from decimal import Decimal, InvalidOperation

MCP_VERSION = '2025-11-25'

MCP_MAX_BYTES = 8192

MCP_INPUT_SCHEMA = {
    'type':'object',
    'properties':{'a':{'type':'integer','minimum':0,'maximum':1000},
                  'b':{'type':'integer','minimum':0,'maximum':1000}},
    'required':['a','b'], 'additionalProperties':False}

MCP_OUTPUT_SCHEMA = {
    'type':'object',
    'properties':{'value':{'type':'integer','minimum':0,'maximum':1000000},
                  'status':{'type':'string','enum':['ok','invalid-arguments','limit-exceeded']}},
    'required':['value','status'], 'additionalProperties':False}

def _mcp_integer(value, low, high):
    if type(value) not in (int,float) or not low <= value <= high or not math.isfinite(value) or value != int(value):
        raise ValueError('Een begrensde JSON-integerwaarde vereist; bool is geen getal')
    return int(value)

def mcp_id_key(identifier):
    """Numeric 1 and 1.0 correlate; string '1' is a different ID."""
    if type(identifier) is str:
        if not re.fullmatch(r'[A-Za-z0-9_-]{1,40}',identifier):
            raise ValueError('ID-tekst buiten de oefengrens')
        return ('string',identifier)
    return ('number',_mcp_integer(identifier,-1000000,1000000))

def _mcp_json_tree(value):
    remaining = [512]
    active = set()
    def visit(item, depth):
        remaining[0] -= 1
        if remaining[0] < 0 or depth > 8:
            raise ValueError('JSON-boom buiten de oefengrens')
        if item is None or type(item) is bool:
            return
        if type(item) is str:
            if len(item) > 4000:
                raise ValueError('JSON-tekst te lang')
            return
        if type(item) in (int,float):
            if abs(item) > 1000000000 or not math.isfinite(item):
                raise ValueError('JSON-getal buiten de oefengrens')
            return
        if type(item) not in (dict,list) or id(item) in active:
            raise ValueError('Alleen eindige JSON-bomen')
        active.add(id(item))
        if type(item) is dict:
            if len(item) > 32 or any(type(key) is not str or len(key) > 128 for key in item):
                raise ValueError('JSON-object buiten de oefengrens')
            for key, child in item.items():
                visit(key,depth+1); visit(child,depth+1)
        else:
            if len(item) > 64:
                raise ValueError('JSON-lijst te lang')
            for child in item:
                visit(child,depth+1)
        active.remove(id(item))
    visit(value,0)

def _mcp_json_bytes(value):
    _mcp_json_tree(value)
    try:
        payload = json.dumps(value,ensure_ascii=False,allow_nan=False,separators=(',',':')).encode('utf-8')
    except (ValueError,UnicodeError) as error:
        raise ValueError('Geen geldige UTF-8/JSON') from error
    if len(payload) > MCP_MAX_BYTES:
        raise ValueError('Pakket groter dan de oefengrens van8192bytes')
    return payload

def _mcp_parse_json(text):
    def pairs(items):
        result = {}
        for key,value in items:
            if key in result:
                raise ValueError('Dubbele JSON-objectsleutel')
            result[key] = value
        return result
    def constant(value):
        raise ValueError('Geen NaN of Infinity in JSON')
    def decimal_number(token):
        # Reject a lossy wire conversion rather than accepting a fractional ID
        # or schema value as an integer after binary-float rounding/underflow.
        try:
            exact = Decimal(token)
            number = float(token)
            if not math.isfinite(number) or Decimal(str(number)) != exact:
                raise ValueError('Wiregetal verliest decimale precisie in de oefenruntime')
            return number
        except (InvalidOperation,OverflowError,ValueError) as error:
            raise ValueError('Wiregetal buiten de precieze oefenvorm') from error
    try:
        return json.loads(text,object_pairs_hook=pairs,parse_constant=constant,parse_float=decimal_number)
    except (ValueError,RecursionError) as error:
        raise ValueError('Ongeldige JSON') from error

def mcp_envelope(message):
    """Validate and copy the selected envelope shapes; no method authority."""
    clean = _mcp_parse_json(_mcp_json_bytes(message).decode('utf-8'))
    if type(clean) is not dict or clean.get('jsonrpc') != '2.0':
        raise ValueError('Eén JSON-RPC2.0-object vereist')
    fields = set(clean)
    if 'method' in clean:
        if not {'jsonrpc','method'} <= fields <= {'jsonrpc','method','id','params'}:
            raise ValueError('Exact request-/notification-envelope')
        if type(clean['method']) is not str or not re.fullmatch(r'[A-Za-z][A-Za-z0-9_/.-]{0,79}',clean['method']):
            raise ValueError('Methode buiten de oefengrens')
        if 'params' in clean and type(clean['params']) is not dict:
            raise ValueError('MCP-params is een object')
        if '_meta' in clean.get('params',{}):
            validate_mcp_meta(clean['params']['_meta'],request=True)
        if 'id' in clean:
            mcp_id_key(clean['id'])
    elif 'result' in clean:
        if fields != {'jsonrpc','id','result'} or type(clean['result']) is not dict:
            raise ValueError('Result-response heeft ID en objectresultaat')
        mcp_id_key(clean['id'])
        if '_meta' in clean['result']:
            validate_mcp_meta(clean['result']['_meta'])
    elif 'error' in clean:
        if not {'jsonrpc','error'} <= fields <= {'jsonrpc','id','error'}:
            raise ValueError('Exact error-envelope vereist')
        if 'id' in clean:
            mcp_id_key(clean['id'])
        error = clean['error']
        if type(error) is not dict or not {'code','message'} <= set(error) <= {'code','message','data'}:
            raise ValueError('Error heeft code/message en eventueel data')
        _mcp_integer(error['code'],-1000000,1000000)
        if type(error['message']) is not str or not error['message'].strip() or len(error['message']) > 2000:
            raise ValueError('Begrensde foutmelding vereist')
    else:
        raise ValueError('Request, notification, result of error vereist')
    return clean

def mcp_message_kind(message):
    clean = mcp_envelope(message)
    if 'method' in clean:
        return 'request' if 'id' in clean else 'notification'
    return 'result' if 'result' in clean else 'error'

def encode_mcp_packet(message):
    return _mcp_json_bytes(mcp_envelope(message))

def decode_mcp_packet(payload):
    if type(payload) is not bytes or not 1 <= len(payload) <= MCP_MAX_BYTES:
        raise ValueError('Eén begrensd bytespakket vereist')
    try:
        text = payload.decode('utf-8')
    except UnicodeError as error:
        raise ValueError('Pakket is geen UTF-8') from error
    return mcp_envelope(_mcp_parse_json(text))

def validate_fixed_mcp_schema(schema,value):
    """Only these two exact classroom schemas, never arbitrary JSON Schema."""
    try:
        signature = json.dumps(schema,sort_keys=True,allow_nan=False,separators=(',',':'))
    except (TypeError,ValueError,RecursionError) as error:
        raise ValueError('Geen ondersteund oefenschema') from error
    input_signature = json.dumps(MCP_INPUT_SCHEMA,sort_keys=True,separators=(',',':'))
    output_signature = json.dumps(MCP_OUTPUT_SCHEMA,sort_keys=True,separators=(',',':'))
    if type(value) is not dict:
        raise ValueError('Schemas hebben een object als wortel')
    if signature == input_signature:
        if set(value) != {'a','b'}:
            raise ValueError('Exact a en b vereist')
        return {key:_mcp_integer(value[key],0,1000) for key in ['a','b']}
    if signature == output_signature:
        if set(value) != {'value','status'} or type(value['status']) is not str or value['status'] not in ['ok','invalid-arguments','limit-exceeded']:
            raise ValueError('Exact value en bekende status vereist')
        return {'value':_mcp_integer(value['value'],0,1000000),'status':value['status']}
    raise ValueError('Onbekend dialect/keyword/schema: niet gevalideerd')

def validate_mcp_peer(info):
    if type(info) is not dict or not {'name','version'} <= set(info):
        raise ValueError('Peer heeft name en version')
    if any(type(info[key]) is not str or not info[key].strip() or len(info[key]) > 80 for key in ['name','version']):
        raise ValueError('Begrensde peeridentificatie vereist')
    info = _mcp_parse_json(_mcp_json_bytes(info).decode('utf-8'))
    for key in ['title','description','websiteUrl']:
        if key in info and (type(info[key]) is not str or len(info[key]) > 2000):
            raise ValueError('Optionele peerbeschrijving is begrensde tekst')
    if 'icons' in info:
        if type(info['icons']) is not list or len(info['icons']) > 16:
            raise ValueError('Begrensde iconmetadata vereist; er wordt niets opgehaald')
        for icon in info['icons']:
            if type(icon) is not dict or 'src' not in icon or type(icon['src']) is not str:
                raise ValueError('Iconmetadata heeft een tekstbron')
            if 'mimeType' in icon and type(icon['mimeType']) is not str:
                raise ValueError('MIME-metadata is tekst')
            if 'sizes' in icon and (type(icon['sizes']) is not list or any(type(size) is not str for size in icon['sizes'])):
                raise ValueError('Icongroottes zijn tekstwaarden')
            if 'theme' in icon and icon['theme'] not in ['light','dark']:
                raise ValueError('Bekend iconthema vereist')
    return info

def validate_mcp_capabilities(capabilities,role='server'):
    if type(capabilities) is not dict or any(type(value) is not dict for value in capabilities.values()):
        raise ValueError('Capabilitynamen verwijzen naar objecten, ook een leeg object')
    capabilities = _mcp_parse_json(_mcp_json_bytes(capabilities).decode('utf-8'))
    client = {'experimental':'object-map','roots':{'listChanged':'bool'},
              'sampling':{'context':'object','tools':'object'},'elicitation':{'form':'object','url':'object'},
              'tasks':{'list':'object','cancel':'object','requests':{
                  'sampling':{'createMessage':'object'},'elicitation':{'create':'object'}}}}
    server = {'experimental':'object-map','logging':'object','completions':'object',
              'prompts':{'listChanged':'bool'},'resources':{'subscribe':'bool','listChanged':'bool'},
              'tools':{'listChanged':'bool'},
              'tasks':{'list':'object','cancel':'object','requests':{'tools':{'call':'object'}}}}
    if role not in ['client','server']:
        raise ValueError('Client- of servercapabilities vereist')
    def check(data,shape):
        for key,kind in shape.items():
            if key not in data:
                continue
            value = data[key]
            if kind == 'bool':
                if type(value) is not bool:
                    raise ValueError('Bekende capabilityvlag is een boolean')
            elif type(value) is not dict:
                raise ValueError('Bekende capabilitystructuur is een object')
            elif kind == 'object-map':
                if any(type(item) is not dict for item in value.values()):
                    raise ValueError('Experimentele capabilitywaarden zijn objecten')
            elif type(kind) is dict:
                check(value,kind)
    check(capabilities,client if role == 'client' else server)
    # Unknown extensions remain bounded inert data; no automatic use/authority.
    return capabilities

def validate_mcp_meta(metadata,request=False):
    if type(metadata) is not dict:
        raise ValueError('Metadata is een object')
    metadata = _mcp_parse_json(_mcp_json_bytes(metadata).decode('utf-8'))
    if request and 'progressToken' in metadata and type(metadata['progressToken']) not in (str,int,float):
        raise ValueError('Een voortgangstoken is tekst of een getal, geen boolean')
    return metadata

def validate_mcp_versions(versions):
    if type(versions) not in (list,tuple) or not 1 <= len(versions) <= 4:
        raise ValueError('1–4 eigen ondersteunde versies vereist')
    if any(type(value) is not str or not value.strip() or len(value) > 40 for value in versions) or len(set(versions)) != len(versions):
        raise ValueError('Unieke begrensde versieteksten vereist')
    return list(versions)

def validate_mcp_initialize(request):
    request = mcp_envelope(request)
    if mcp_message_kind(request) != 'request' or request['method'] != 'initialize':
        raise ValueError('Initialize-request vereist')
    params = request.get('params')
    required = {'protocolVersion','capabilities','clientInfo'}
    if type(params) is not dict or not required <= set(params) <= required | {'_meta'}:
        raise ValueError('Gekozen initialize-params vereist')
    validate_mcp_versions([params['protocolVersion']])
    validate_mcp_capabilities(params['capabilities'],role='client')
    validate_mcp_peer(params['clientInfo'])
    if '_meta' in params:
        validate_mcp_meta(params['_meta'],request=True)
    return request

def mcp_tool_definition():
    return {'name':'multiply','description':'Vermenigvuldig twee eigen oefengetallen; productlimiet5000.',
            'inputSchema':copy.deepcopy(MCP_INPUT_SCHEMA),'outputSchema':copy.deepcopy(MCP_OUTPUT_SCHEMA),
            'annotations':{'readOnlyHint':True,'destructiveHint':False}}

def mcp_tool_result(value,status='ok'):
    structured = validate_fixed_mcp_schema(MCP_OUTPUT_SCHEMA,{'value':value,'status':status})
    if status != 'ok' and structured['value'] != 0:
        raise ValueError('Eigen foutresultaat gebruikt waarde0')
    return {'content':[{'type':'text','text':json.dumps(structured,separators=(',',':'))}],
            'structuredContent':structured,'isError':status != 'ok'}

def validate_mcp_tool_result(result):
    """Selected text+structured result, including the tool's error branch."""
    if type(result) is not dict or not {'content','structuredContent'} <= set(result) <= {'content','structuredContent','isError'}:
        raise ValueError('Exact ondersteund toolresultaat vereist')
    structured = validate_fixed_mcp_schema(MCP_OUTPUT_SCHEMA,result['structuredContent'])
    is_error = result.get('isError',False)
    if type(is_error) is not bool or is_error != (structured['status'] != 'ok'):
        raise ValueError('isError en eigen status spreken elkaar tegen')
    if is_error and structured['value'] != 0:
        raise ValueError('Eigen foutresultaat gebruikt waarde0')
    content = result['content']
    if type(content) is not list or len(content) != 1 or type(content[0]) is not dict or set(content[0]) != {'type','text'}:
        raise ValueError('Eén gegeven tekstkopie vereist')
    if content[0]['type'] != 'text' or type(content[0]['text']) is not str or len(content[0]['text']) > 4000:
        raise ValueError('Begrensde tekstkopie vereist')
    mirrored = validate_fixed_mcp_schema(MCP_OUTPUT_SCHEMA,_mcp_parse_json(content[0]['text']))
    if mirrored != structured:
        raise ValueError('Tekst en gestructureerde data verschillen')
    return copy.deepcopy({**result,'isError':is_error,'structuredContent':structured})

def validate_mcp_call(params):
    if type(params) is not dict:
        raise ValueError('Tools/call-params is een object')
    params = _mcp_parse_json(_mcp_json_bytes(params).decode('utf-8'))
    if not {'name'} <= set(params) <= {'name','arguments','_meta'}:
        raise ValueError('Name en optioneel arguments/metadata vereist')
    if type(params['name']) is not str or not re.fullmatch(r'[A-Za-z0-9_.-]{1,128}',params['name']):
        raise ValueError('Toolnaam buiten de oefengrens')
    arguments = params.get('arguments',{})
    if type(arguments) is not dict or ('_meta' in params and type(params['_meta']) is not dict):
        raise ValueError('Arguments en metadata zijn objecten')
    if '_meta' in params:
        validate_mcp_meta(params['_meta'],request=True)
    return {**params,'arguments':arguments}

def execute_mcp_multiply(arguments):
    try:
        clean = validate_fixed_mcp_schema(MCP_INPUT_SCHEMA,arguments)
    except ValueError:
        return mcp_tool_result(0,'invalid-arguments')
    product = clean['a'] * clean['b']
    if product > 5000:
        return mcp_tool_result(0,'limit-exceeded')
    return mcp_tool_result(product)

def dispatch_mcp_tool(message,tools_available=True):
    if type(tools_available) is not bool:
        raise ValueError('Toolbeschikbaarheid is een expliciete hostkeuze')
    message = mcp_envelope(message)
    kind = mcp_message_kind(message)
    if kind == 'notification':
        return None
    if kind != 'request':
        raise ValueError('Response hoort bij clientcorrelatie, niet bij tooldispatch')
    def error(code,text):
        return {'jsonrpc':'2.0','id':message['id'],'error':{'code':code,'message':text}}
    def result(value):
        return {'jsonrpc':'2.0','id':message['id'],'result':copy.deepcopy(value)}
    if message['method'] not in ['tools/list','tools/call']:
        return error(-32601,'Method not found')
    if not tools_available:
        return error(-32601,'Tools capability unavailable')
    params = message.get('params',{})
    if message['method'] == 'tools/list':
        if set(params) - {'_meta'} or ('_meta' in params and type(params['_meta']) is not dict):
            return error(-32602,'Unsupported list parameters')
        return result({'tools':[mcp_tool_definition()]})
    try:
        call = validate_mcp_call(params)
    except ValueError:
        return error(-32602,'Invalid tool call parameters')
    if call['name'] != 'multiply':
        return error(-32602,'Unknown tool')
    return result(execute_mcp_multiply(call['arguments']))

request = {'jsonrpc':'2.0','id':0,'method':'tools/call','params':{'name':'multiply','arguments':{'a':2,'b':3}}}
response = dispatch_mcp_tool(request)
result = validate_mcp_tool_result(response['result'])
print(result['structuredContent']['value'], result['isError'])

Voer de code uit en verklaar voor jezelf waarom elke stap werkt.