"""Same explicit MCP schemas for Bridge and NEXUS Gateway. No generic JS, arbitrary RPC operation, raw URL request or secret field is exposed. """ from __future__ import annotations import json from typing import Any from web_guarded import GuardError, Scope, parse_spec WEB_OPERATIONS = frozenset({ 'browser_web_observe','browser_web_action','browser_web_wait','browser_web_receipt', 'browser_web_cancel','browser_web_download_read', }) _SCOPE_FIELDS = {'task_id','worker_id','generation','session_id','lease_id'} _FIELDS = { 'browser_web_observe': _SCOPE_FIELDS | {'page_id','frame_id','max_nodes','offset'}, 'browser_web_action': _SCOPE_FIELDS | {'observation_id','ref_id','request_id','action','spec_json','user_confirmed','sensitive_confirmed'}, 'browser_web_wait': _SCOPE_FIELDS | {'observation_id','ref_id','condition','expected_json','timeout_ms'}, 'browser_web_receipt': _SCOPE_FIELDS | {'request_id'}, 'browser_web_cancel': _SCOPE_FIELDS | {'request_id','user_confirmed'}, 'browser_web_download_read': _SCOPE_FIELDS | {'download_id','offset','max_bytes'}, } def _build_upload_spec(files, inline_file_name: str, inline_content_base64: str, inline_expected_sha256: str) -> dict: inline_supplied = any(value != '' for value in (inline_file_name, inline_content_base64, inline_expected_sha256)) fileparam_supplied = files is not None if int(fileparam_supplied) + int(inline_supplied) != 1: raise GuardError('UPLOAD_ROUTE_EXCLUSIVITY_REQUIRED') if fileparam_supplied: return {'files': files} if not all(isinstance(value, str) and value for value in (inline_file_name, inline_content_base64, inline_expected_sha256)): raise GuardError('UPLOAD_INLINE_FIELDS_REQUIRED') return {'inline_file': { 'file_name': inline_file_name, 'content_base64': inline_content_base64, 'expected_sha256': inline_expected_sha256, }} async def dispatch_bridge_web(service, owner: str, operation: str, payload: dict) -> dict: if operation not in WEB_OPERATIONS: raise GuardError('WEB_OPERATION_NOT_ALLOWED') if not isinstance(payload,dict) or set(payload) - _FIELDS[operation] or _SCOPE_FIELDS - set(payload): raise GuardError('WEB_PAYLOAD_FIELDS_INVALID') scope=Scope(owner=owner,**{k:payload[k] for k in _SCOPE_FIELDS}) scope.validate() if operation=='browser_web_observe': return await service.observe(scope,page_id=payload.get('page_id',''), frame_id=payload.get('frame_id',''),max_nodes=payload.get('max_nodes',60),offset=payload.get('offset',0)) if operation=='browser_web_action': return await service.action(scope,observation_id=payload['observation_id'],ref_id=payload.get('ref_id',''), request_id=payload['request_id'],action=payload['action'],spec=parse_spec(payload.get('spec_json','{}')), user_confirmed=payload.get('user_confirmed',False),sensitive_confirmed=payload.get('sensitive_confirmed',False)) if operation=='browser_web_wait': spec=parse_spec(payload.get('expected_json','{}')) if set(spec)!={'expected'}: raise GuardError('WAIT_EXPECTED_OBJECT_REQUIRED') return await service.wait(scope,observation_id=payload['observation_id'],ref_id=payload['ref_id'], condition=payload['condition'],expected=spec['expected'],timeout_ms=payload.get('timeout_ms',1500)) if operation=='browser_web_receipt': return service.receipt(scope,payload['request_id']) if operation=='browser_web_cancel': return await service.cancel(scope,request_id=payload['request_id'], user_confirmed=payload.get('user_confirmed',False)) return await service.download_read(scope,download_id=payload['download_id'], offset=payload.get('offset',0),max_bytes=payload.get('max_bytes',65536)) def _register(server, call, prefix: str): # Lazy import keeps pure safety tests independent of an installed MCP runtime. from mcp.types import ToolAnnotations read=ToolAnnotations(readOnlyHint=True, destructiveHint=False, idempotentHint=False, openWorldHint=True) write=ToolAnnotations(readOnlyHint=False, destructiveHint=False, idempotentHint=False, openWorldHint=True) @server.tool(name=prefix+'observe',annotations=read,description=( 'Observe controls on one exact leased browser page/frame. Returns fresh opaque refs, no field values. ' 'Returns up to 400 controls. Offset pages through long control lists, up to index 10000. Each slice is fresh and invalidates prior refs. ' 'No automatic popup switch, no Windows input. task/worker/generation are checked against server assignment.')) async def observe(session_id: str,lease_id: str,task_id: str,worker_id: str,generation: int, page_id: str='',frame_id: str='',max_nodes: int=60,offset: int=0): return await call('browser_web_observe',dict(session_id=session_id,lease_id=lease_id,task_id=task_id, worker_id=worker_id,generation=generation,page_id=page_id,frame_id=frame_id,max_nodes=max_nodes,offset=offset)) @server.tool(name=prefix+'action',annotations=write,description=( 'One reviewed leased browser action on a fresh observation: semantic click/hover/fill/select/check/scroll/select_page, ' 'including ARIA buttons, links, tabs, menuitems and treeitems. ' 'navigate, back, upload or public download. Scroll may target a fresh scrollable-container ref and axis x/y. ' 'Canvas remains disabled. Never submit or inject OS input. ' 'Persisted request_id prevents replay. On transport loss read receipt with the SAME binding/id.')) async def action(session_id: str,lease_id: str,task_id: str,worker_id: str,generation: int, request_id: str,observation_id: str,action: str,ref_id: str='',spec_json: str='{}', user_confirmed: bool=False,sensitive_confirmed: bool=False): return await call('browser_web_action',dict(session_id=session_id,lease_id=lease_id,task_id=task_id, worker_id=worker_id,generation=generation,request_id=request_id,observation_id=observation_id, action=action,ref_id=ref_id,spec_json=spec_json,user_confirmed=user_confirmed, sensitive_confirmed=sensitive_confirmed)) @server.tool(name=prefix+'wait',annotations=read,description=( 'Bounded read-only match on an existing observation/ref. No repeated action, no raw value returned. ' 'Supports visible/checked/input_equals/selected_equals/expanded/aria_selected/enabled/label_equals, timeout 0..3000 ms.')) async def wait(session_id: str,lease_id: str,task_id: str,worker_id: str,generation: int, observation_id: str,ref_id: str,condition: str,expected_json: str,timeout_ms: int=1500): return await call('browser_web_wait',dict(session_id=session_id,lease_id=lease_id,task_id=task_id, worker_id=worker_id,generation=generation,observation_id=observation_id,ref_id=ref_id, condition=condition,expected_json=expected_json,timeout_ms=timeout_ms)) @server.tool(name=prefix+'receipt',annotations=read,description=( 'Read the exact persisted action outcome, including PENDING/UNKNOWN, without replay. ' 'NOT_FOUND is NOT permission to repeat an uncertain write. Works without a live page.')) async def receipt(session_id: str,lease_id: str,task_id: str,worker_id: str,generation: int,request_id: str): return await call('browser_web_receipt',dict(session_id=session_id,lease_id=lease_id,task_id=task_id, worker_id=worker_id,generation=generation,request_id=request_id)) @server.tool(name=prefix+'cancel',annotations=write,description=( 'Cancel one active guarded action by its exact request_id. Cancellation never starts or repeats ' 'the action. Read the same receipt afterwards.')) async def cancel(session_id: str,lease_id: str,task_id: str,worker_id: str,generation: int, request_id: str,user_confirmed: bool=False): return await call('browser_web_cancel',dict(session_id=session_id,lease_id=lease_id,task_id=task_id, worker_id=worker_id,generation=generation,request_id=request_id,user_confirmed=user_confirmed)) @server.tool(name=prefix+'download_read',annotations=read,description=( 'Read one bounded base64 chunk from a guarded public download. Exact session, lease and download_id ' 'are required. The raw chunk limit is 131072 bytes and session close removes the temporary file.')) async def download_read(session_id: str,lease_id: str,task_id: str,worker_id: str,generation: int, download_id: str,offset: int=0,max_bytes: int=65536): return await call('browser_web_download_read',dict(session_id=session_id,lease_id=lease_id, task_id=task_id,worker_id=worker_id,generation=generation,download_id=download_id, offset=offset,max_bytes=max_bytes)) @server.tool(name=prefix+'upload',annotations=write,meta={'openai/fileParams':['files']},description=( 'Attach exactly one file to an observed file input without submitting. Backward-compatible files uses ' 'ChatGPT fileParams. If that platform conversion is unavailable, a confirmed small file may instead use ' 'inline_file_name + inline_content_base64 + inline_expected_sha256. Exactly one route is required; inline ' 'content is hash-bound, limited to 12 KiB, never returned in receipts, and executable payloads are blocked.')) async def upload(session_id: str,lease_id: str,task_id: str,worker_id: str,generation: int, request_id: str,observation_id: str,ref_id: str, files: list[dict[str,str]] | None=None,inline_file_name: str='', inline_content_base64: str='',inline_expected_sha256: str='', user_confirmed: bool=False): spec=_build_upload_spec(files,inline_file_name,inline_content_base64,inline_expected_sha256) spec_json=json.dumps(spec,ensure_ascii=True,separators=(',',':'),allow_nan=False) return await call('browser_web_action',dict(session_id=session_id,lease_id=lease_id,task_id=task_id, worker_id=worker_id,generation=generation,request_id=request_id,observation_id=observation_id, action='upload',ref_id=ref_id,spec_json=spec_json,user_confirmed=user_confirmed, sensitive_confirmed=True)) def register_bridge_web_tools(server, service, owner_get): async def call(operation,payload): # Do not catch authentication errors or substitute another owner's session. owner=owner_get() return await dispatch_bridge_web(service,owner,operation,payload) _register(server,call,'browser_web_') def register_gateway_web_tools(server, relay, render): async def call(operation,payload): # Existing relay owns auth, 429, cooldown and transport uncertainty. # No retry loop, alternate route or low-level browser fallback here. return render(await relay(operation,payload)) _register(server,call,'nexus_browser_web_')