Voeg workerresultaten pas samen na controle

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_coordination_task(task):
    if type(task) is not dict or set(task)!={'id','version','values'}:
        raise ValueError('Exacte eigen taak met id/version/values vereist')
    if type(task['id']) is not str or not re.fullmatch(r'[a-z][a-z0-9-]{0,31}',task['id']):
        raise ValueError('Begrensde eigen taak-ID vereist')
    if type(task['values']) is not list or not 1<=len(task['values'])<=8:
        raise ValueError('1–8 eigen getallen vereist')
    return {'id':task['id'],'version':_mcp_integer(task['version'],0,1000000),
            'values':[_mcp_integer(value,0,1000) for value in task['values']]}

def run_classroom_subtask(subtask):
    if type(subtask) is not dict or set(subtask)!={'subtask_id','task_id','source_version','values'} or subtask['subtask_id'] not in ['count','sum']:
        raise ValueError('Bekende eigen workeropdracht vereist')
    task=validate_coordination_task({'id':subtask['task_id'],'version':subtask['source_version'],'values':subtask['values']})
    value=len(task['values']) if subtask['subtask_id']=='count' else sum(task['values'])
    return {'subtask_id':subtask['subtask_id'],'task_id':task['id'],'source_version':task['version'],
            'status':'ok','value':value,'reason':None}

def create_subtasks(task):
    task=validate_coordination_task(task)
    return [{'subtask_id':name,'task_id':task['id'],'source_version':task['version'],
             'values':list(task['values'])} for name in ['count','sum']]

def check_subtask_result(task,result):
    task=validate_coordination_task(task)
    result=_mcp_parse_json(_mcp_json_bytes(result).decode('utf-8'))
    if type(result) is not dict or set(result)!={'subtask_id','task_id','source_version','status','value','reason'}:
        raise ValueError('Exact eigen resultaatcontract vereist')
    if type(result['subtask_id']) is not str or result['subtask_id'] not in ['count','sum'] or result['task_id']!=task['id'] or _mcp_integer(result['source_version'],0,1000000)!=task['version']:
        raise ValueError('Resultaat hoort bij deze subtaak, taak en bronversie')
    result['source_version']=task['version']
    if result['status']=='ok':
        expected=len(task['values']) if result['subtask_id']=='count' else sum(task['values'])
        result['value']=_mcp_integer(result['value'],0,8000)
        if result['value']!=expected or result['reason'] is not None:
            raise ValueError('Eigen deterministische uitkomst klopt niet')
    elif result['status']=='error':
        if result['value'] is not None or type(result['reason']) is not str or not result['reason'].strip() or len(result['reason'])>200:
            raise ValueError('Workerfout heeft geen waarde en wel een begrensde reden')
    else:
        raise ValueError('Bekende workerstatus vereist')
    return result

def join_subtasks(task,results):
    task=validate_coordination_task(task)
    if type(results) is not list or len(results)>4:
        raise ValueError('0–4 ontvangen toepassingsresultaten vereist')
    requests=create_subtasks(task)
    request_packets=[_mcp_json_bytes(item) for item in requests]
    reply_packets=[_mcp_json_bytes(item) for item in results]
    base={'average':None,'received':len(results),'unique':0,'results':[],
          'messages':len(request_packets)+len(reply_packets),
          'bytes':sum(len(payload) for payload in request_packets+reply_packets),
          'analytical_plan':{'work_units':7,'critical_path_units':5,'direct_worker_messages':0}}
    try:
        checked=[check_subtask_result(task,result) for result in results]
    except ValueError:
        return {**base,'status':'rejected','reason':'invalid-result'}
    by_id={};conflict=False
    for result in checked:
        name=result['subtask_id']
        if name in by_id and by_id[name]!=result:
            conflict=True
        else:
            by_id[name]=result
    accepted=[by_id[name] for name in ['count','sum'] if name in by_id]
    base.update(unique=len(by_id),results=copy.deepcopy(accepted))
    if conflict:
        return {**base,'status':'conflict','reason':'conflicting-subtask'}
    if any(result['status']=='error' for result in accepted):
        return {**base,'status':'failed','reason':'worker-error'}
    if len(by_id)!=2:
        return {**base,'status':'partial','reason':'missing-subtask'}
    return {**base,'status':'ok','reason':None,'average':by_id['sum']['value']/by_id['count']['value']}

def run_coordination_demo(task):
    task=validate_coordination_task(task)
    request_packets=[];reply_packets=[];results=[]
    for subtask in create_subtasks(task):
        request_payload=_mcp_json_bytes(subtask)
        request_packets.append(request_payload)
        received=_mcp_parse_json(request_payload.decode('utf-8'))
        reply=run_classroom_subtask(received)
        reply_payload=_mcp_json_bytes(reply)
        reply_packets.append(reply_payload)
        results.append(_mcp_parse_json(reply_payload.decode('utf-8')))
    report=join_subtasks(task,results)
    return {'request_packets':request_packets,'reply_packets':reply_packets,'report':report}

task = {'id':'calc-a','version':3,'values':[2,3,4]}
coordination = run_coordination_demo(task)
report = coordination['report']
print(report['status'],report['average'])
print(report['messages'],report['bytes'])

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