Begin een MCP-verbinding met heldere afspraken

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 server_initialize(request):
    request = validate_mcp_initialize(request)
    # This server supports one fixed edition. A different offer receives that edition.
    return {'jsonrpc':'2.0','id':request['id'],'result':{
        'protocolVersion':MCP_VERSION,'capabilities':{'tools':{}},
        'serverInfo':{'name':'ai-lab-server','version':'1.0.0'}}}

def client_initialize(request,response,supported_versions=(MCP_VERSION,)):
    request = validate_mcp_initialize(request)
    supported = validate_mcp_versions(supported_versions)
    if request['params']['protocolVersion'] not in supported:
        raise ValueError('Client moet de aangeboden versie zelf ondersteunen')
    def outcome(phase,reason=None,version=None,capabilities=None,outgoing=None):
        return {'phase':phase,'reason':reason,'version':version,
                'capabilities':{} if capabilities is None else copy.deepcopy(capabilities),
                'outgoing':[] if outgoing is None else copy.deepcopy(outgoing)}
    try:
        response = mcp_envelope(response)
    except ValueError:
        return outcome('closed','invalid-response')
    kind = mcp_message_kind(response)
    if kind not in ['result','error']:
        return outcome('initializing','ignored-non-response')
    if 'id' not in response:
        return outcome('closed','uncorrelated-error')
    if mcp_id_key(response['id']) != mcp_id_key(request['id']):
        return outcome('initializing','unmatched-response')
    if kind == 'error':
        return outcome('closed','initialize-error')
    result = response['result']
    try:
        if not {'protocolVersion','capabilities','serverInfo'} <= set(result):
            raise ValueError('Initialize-resultaat mist velden')
        validate_mcp_versions([result['protocolVersion']])
        capabilities = validate_mcp_capabilities(result['capabilities'])
        validate_mcp_peer(result['serverInfo'])
        if 'instructions' in result and (type(result['instructions']) is not str or len(result['instructions']) > 2000):
            raise ValueError('Optionele instructietekst is begrensde data')
    except ValueError:
        return outcome('closed','invalid-initialize-result')
    if result['protocolVersion'] not in supported:
        return outcome('closed','version-mismatch')
    notification = {'jsonrpc':'2.0','method':'notifications/initialized'}
    return outcome('ready',version=result['protocolVersion'],capabilities=capabilities,outgoing=[notification])

def run_mcp_handshake(events,supported_versions=(MCP_VERSION,)):
    supported = validate_mcp_versions(supported_versions)
    if type(events) is not list or not 1 <= len(events) <= 8:
        raise ValueError('1–8 eigen lifecycle-events vereist')
    state = {'phase':'new','reason':None,'version':None,'capabilities':{}}
    outgoing = []; request = None
    sent_bytes = received_bytes = received_packets = received_messages = ignored = 0
    for event in events:
        if type(event) is not dict or state['phase'] == 'closed':
            raise ValueError('Event is een dictionary; gesloten transport verwerkt niets meer')
        kind = event.get('kind')
        if kind == 'start':
            if set(event) != {'kind','request'} or state['phase'] != 'new':
                raise ValueError('Start precies eenmaal vanuit nieuw')
            request = validate_mcp_initialize(event['request'])
            if request['params']['protocolVersion'] not in supported:
                raise ValueError('Client biedt eigen ondersteunde versie aan')
            payload = encode_mcp_packet(request)
            outgoing.append(decode_mcp_packet(payload)); sent_bytes += len(payload)
            state['phase'] = 'initializing'
        elif kind == 'reply':
            if set(event) != {'kind','packet'} or state['phase'] != 'initializing' or type(event['packet']) is not bytes or len(event['packet']) > MCP_MAX_BYTES:
                raise ValueError('Begrensd antwoordpakket tijdens initialisatie vereist')
            received_packets += 1; received_bytes += len(event['packet'])
            try:
                response = decode_mcp_packet(event['packet'])
            except ValueError:
                state = {'phase':'closed','reason':'invalid-response','version':None,'capabilities':{}}
                continue
            received_messages += 1
            result = client_initialize(request,response,supported)
            state = {key:copy.deepcopy(result[key]) for key in ['phase','reason','version','capabilities']}
            if state['phase'] == 'initializing':
                ignored += 1
            for message in result['outgoing']:
                payload = encode_mcp_packet(message)
                outgoing.append(decode_mcp_packet(payload)); sent_bytes += len(payload)
        elif kind == 'timeout':
            if set(event) != {'kind'} or state['phase'] != 'initializing':
                raise ValueError('Initialisatietimeout hoort bij wachten op initialize')
            state = {'phase':'closed','reason':'initialization-timeout','version':None,'capabilities':{}}
        elif kind == 'close':
            if set(event) != {'kind'}:
                raise ValueError('Close-event heeft geen protocolpayload')
            state = {'phase':'closed','reason':'transport-closed','version':None,'capabilities':{}}
        else:
            raise ValueError('Onbekend lifecycle-event')
    return {**state,'outgoing':outgoing,'sent_messages':len(outgoing),'received_packets':received_packets,'received_messages':received_messages,
            'sent_bytes':sent_bytes,'received_bytes':received_bytes,'ignored_responses':ignored}

request = {'jsonrpc':'2.0','id':0,'method':'initialize','params':{
    'protocolVersion':'2025-11-25','capabilities':{},'clientInfo':{'name':'ai-lab-client','version':'1.0.0'}}}
response = server_initialize(request)
events = [{'kind':'start','request':request},{'kind':'reply','packet':encode_mcp_packet(response)}]
handshake = run_mcp_handshake(events)
print(handshake['phase'], handshake['sent_messages']+handshake['received_messages'])
print(handshake['sent_bytes'], handshake['received_bytes'])

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