"""Python app contract: optional providers, safe event defaults, MCP export adapter.""" from abc import ABC, abstractmethod from dataclasses import dataclass import inspect, json, logging, re CONTRACT="ci-python-app/v1" EVENT_CONTRACT="ci-python-app-event/v1" CONTRACT_RESOURCE_META="tech.christian-igel/python-app" VIEW_FORMATS=("phone-format","tablet-hochformat","tablet-querformat","desktop-format") ROLES=("guest","customer","developer","admin") STANDARD_EVENTS=("window-ready","window-state-changed","window-size-changed","window-position-changed","window-role-changed","window-action","window-closed","context-revoked","window-gained-focus","window-lost-focus","view.changed","agent-capabilities-changed") _NAME=re.compile(r"^[a-zA-Z][a-zA-Z0-9_.:-]{0,127}$") LOG=logging.getLogger("ci_app_contract") def _text(v,label,limit=128): if not isinstance(v,str) or not v or len(v)>limit or any(ord(c)<32 for c in v): raise ValueError("Invalid "+label) return v def _json(v): s=json.dumps(v,ensure_ascii=False,allow_nan=False) if len(s.encode())>262144: raise ValueError("Contract payload exceeds limit") return json.loads(s) def view_format(width,height,media_type="screen"): if type(width) is not int or type(height) is not int or not 1<=width<=16384 or not 1<=height<=16384: raise ValueError("Invalid content dimensions") _text(media_type,"media type",80) return VIEW_FORMATS[0] if width<600 else (VIEW_FORMATS[1] if height>width else VIEW_FORMATS[2]) if width<1200 else VIEW_FORMATS[3] @dataclass(frozen=True) class AppInfo: app_id:str name:str version:str revision:int=1 supported_roles:tuple=("customer",) guest_access:bool=False def __post_init__(self): if not _NAME.fullmatch(_text(self.app_id,"app id")): raise ValueError("Invalid app id") _text(self.name,"app name",200); _text(self.version,"version",80) if type(self.revision) is not int or self.revision<1: raise ValueError("Invalid revision") if not isinstance(self.supported_roles,tuple) or not self.supported_roles or len(set(self.supported_roles))!=len(self.supported_roles) or any(r not in ROLES for r in self.supported_roles): raise ValueError("Invalid supported roles") if "guest" in self.supported_roles and not self.guest_access: raise ValueError("Guest support must be explicit") class ToolProvider(ABC): @abstractmethod async def list_tools(self,principal): ... @abstractmethod async def call_tool(self,principal,name,arguments): ... class ResourceProvider(ABC): @abstractmethod async def list_resources(self,principal): ... @abstractmethod async def read_resource(self,principal,uri): ... class WindowUI(ABC): @abstractmethod async def ui_resources(self,principal): ... async def window_actions(self,principal): return [] @abstractmethod async def on_view_changed(self,view): ... class SettingsProvider(ABC): @abstractmethod async def settings_schema(self,principal): ... class StreamProvider(ABC): @abstractmethod async def stream_descriptors(self,principal): ... @abstractmethod async def stop_streams(self,principal,context): ... class EventProvider(ABC): @abstractmethod def event_handlers(self): ... class AgentProvider(ABC): @abstractmethod async def agent_context(self,principal): ... async def agent_tools(self,principal): return [] def agent_capabilities(visible_ids,*,app_allowed=None,user_disabled=()): """Already-authorized visible IDs intersect app restrictions and user opt-outs.""" visible=set(visible_ids); disabled=set(user_disabled) allowed=visible if app_allowed is None else visible.intersection(app_allowed) if any(not isinstance(x,str) or not _NAME.fullmatch(x) for x in visible|disabled|allowed): raise ValueError("Invalid capability id") return sorted(allowed-disabled) def validate_event(raw): keys={"schema","eventId","windowId","instanceId","contextRevision","name","data","correlationId"} if not isinstance(raw,dict) or set(raw)-keys or raw.get("schema")!=EVENT_CONTRACT: raise ValueError("Invalid event envelope") for k in ("eventId","windowId","instanceId"): _text(raw.get(k),k) if not _NAME.fullmatch(_text(raw.get("name"),"event name")): raise ValueError("Invalid event name") if type(raw.get("contextRevision")) is not int or raw["contextRevision"]<0: raise ValueError("Invalid context revision") if "correlationId" in raw: _text(raw["correlationId"],"correlation id") data=raw.get("data",{}); name=raw["name"] if not isinstance(data,dict): raise ValueError("Invalid event data") if name in {"view.changed","window-size-changed"}: fmt=view_format(data.get("width"),data.get("height"),data.get("mediaType","screen")) if name=="view.changed" and data.get("format")!=fmt: raise ValueError("View format does not match dimensions") elif name=="window-state-changed": if data.get("state") not in {"normal","minimized","maximized"}: raise ValueError("Invalid window state") elif name=="window-position-changed": if any(type(data.get(k)) is not int or abs(data[k])>32768 for k in ("x","y")): raise ValueError("Invalid position") elif name=="window-role-changed": if data.get("role") not in ROLES: raise ValueError("Invalid informational role") elif name=="window-action": if not _NAME.fullmatch(_text(data.get("actionId"),"action id")) or not isinstance(data.get("arguments",{}),dict) or "correlationId" not in raw: raise ValueError("Invalid correlated action") elif name=="agent-capabilities-changed": if type(data.get("revision")) is not int or data["revision"]<0: raise ValueError("Invalid agent revision") for key in ("toolIds","mcpIds"): ids=data.get(key,[]) if not isinstance(ids,list) or len(ids)>1000 or any(not isinstance(x,str) or not _NAME.fullmatch(x) for x in ids): raise ValueError("Invalid capability ids") return _json({**raw,"data":data}) class AppBase: """Mandatory identity, description and standard events. Events do not grant rights.""" def __init__(self,info): if not isinstance(info,AppInfo): raise TypeError("AppInfo is required") self.app_info=info; self.current_view=None; self._diagnosed_unknown=set() async def start(self): pass async def close(self): pass async def describe(self,principal=None): tools=await self.list_tools(principal) if isinstance(self,ToolProvider) else [] tools=[x.descriptor() if hasattr(x,"descriptor") else x for x in tools] resources=await self.list_resources(principal) if isinstance(self,ResourceProvider) else [] resources=[x for x in resources if not x.get("uri","").startswith("app-contract://")] ui=await self.ui_resources(principal) if isinstance(self,WindowUI) else [] actions=await self.window_actions(principal) if isinstance(self,WindowUI) else [] ids=[x["id"] for x in actions] if len(ids)!=len(set(ids)) or any(not _NAME.fullmatch(x) for x in ids): raise ValueError("Invalid actions") handlers=self.event_handlers() if isinstance(self,EventProvider) else {} if not isinstance(handlers,dict) or any(not _NAME.fullmatch(x) or not callable(h) for x,h in handlers.items()): raise ValueError("Invalid handlers") providers=(ToolProvider,ResourceProvider,WindowUI,SettingsProvider,StreamProvider,EventProvider,AgentProvider) roles=[r for r in ROLES if r in self.app_info.supported_roles] return _json({"schema":CONTRACT, "app":{"id":self.app_info.app_id,"name":self.app_info.name,"version":self.app_info.version,"revision":self.app_info.revision}, "interfaces":["AppBase"]+[p.__name__ for p in providers if isinstance(self,p)], "types":["app"]+(["tools"] if tools else [])+(["window"] if ui else [])+(["mcp-app"] if any(x.get("mimeType")=="text/html;profile=mcp-app" for x in ui) else []), "supported_roles":roles,"role_selection":len(roles)>1,"guest_access":self.app_info.guest_access, "supported_events":list(dict.fromkeys(STANDARD_EVENTS+tuple(handlers))), "supported_actions":actions,"view_formats":list(VIEW_FORMATS), "view_format_rules":{"coordinate_space":"window-content-css-px","phone_max_exclusive":600,"desktop_min":1200,"tablet_orientation":"height>width","media_type":"declared-rendering-medium"}, "capabilities":{"tools":tools,"resources":resources,"ui":ui, "settings":await self.settings_schema(principal) if isinstance(self,SettingsProvider) else None, "streams":await self.stream_descriptors(principal) if isinstance(self,StreamProvider) else [], "agent_extension":isinstance(self,AgentProvider), "agent_tools":await self.agent_tools(principal) if isinstance(self,AgentProvider) else []}, "agent_policy":{"default":"all-authorized-app-tools","app_can_restrict":True,"window_user_can_disable":True,"revocation":"server-first-before-event"}, "authority":"server-verified-account-app-connection-context"}) async def dispatch_event(self,raw,*,authorize,principal=None): event=validate_event(raw); allowed=authorize(event) if inspect.isawaitable(allowed): allowed=await allowed if allowed is not True: raise PermissionError("Event context denied") name=event["name"]; result={"eventId":event["eventId"],"handled":False} if "correlationId" in event: result["correlationId"]=event["correlationId"] if name=="view.changed": self.current_view=dict(event["data"]) if isinstance(self,WindowUI): await self.on_view_changed(self.current_view) result["handled"]=True elif name=="context-revoked" and isinstance(self,StreamProvider): await self.stop_streams(principal,event); result["handled"]=True elif name=="window-action": actions=await self.window_actions(principal) if isinstance(self,WindowUI) and principal is not None else [] if event["data"]["actionId"] not in {a["id"] for a in actions}: result["diagnostic"]="unsupported-action" return result elif name in STANDARD_EVENTS: result["handled"]=True handlers=self.event_handlers() if isinstance(self,EventProvider) else {} if name in handlers: value=handlers[name](event) if inspect.isawaitable(value): value=await value result.update(handled=True,result=_json(value)) elif name not in STANDARD_EVENTS: result["diagnostic"]="unsupported-event" if name not in self._diagnosed_unknown and len(self._diagnosed_unknown)<128: self._diagnosed_unknown.add(name); LOG.info("unsupported-event app=%s name=%s",self.app_info.app_id,name) return result class McpAppAdapter(AppBase,ToolProvider,ResourceProvider,WindowUI): """Existing MCP exports; no synthetic tools, automatic UI or authorization.""" def __init__(self,info,*,tools,resources,invoke,read,actions=None): super().__init__(info); self._tools=tools; self._resources=resources self._invoke=invoke; self._read=read; self._actions=actions async def _resolve(self,value): return await value if inspect.isawaitable(value) else value async def list_tools(self,principal): return await self._resolve(self._tools(principal)) async def call_tool(self,principal,name,arguments): return await self._resolve(self._invoke(principal,name,arguments)) async def list_resources(self,principal): return await self._resolve(self._resources(principal)) async def read_resource(self,principal,uri): return await self._resolve(self._read(principal,uri)) async def ui_resources(self,principal): resources=await self.list_resources(principal) ui=[x for x in resources if x.get("mimeType")=="text/html;profile=mcp-app" and x.get("uri","").startswith("ui://")] known={x["uri"] for x in ui} for spec in await self.list_tools(principal): d=spec.descriptor() if hasattr(spec,"descriptor") else spec uri=d.get("_meta",{}).get("ui",{}).get("resourceUri") if uri is not None and uri not in known: raise ValueError("Missing MCP UI binding") return ui async def window_actions(self,principal): return [] if self._actions is None else await self._resolve(self._actions(principal)) async def on_view_changed(self,view): # The host delivers the negotiated view to the existing responsive MCP UI. pass