From 643cb65c798841c7f7af34bf342d3eb6471a8010 Mon Sep 17 00:00:00 2001 From: saphid <4596216+saphid@users.noreply.github.com> Date: Mon, 28 Sep 2026 22:21:22 +1000 Subject: [PATCH] Add key-free MCP tools, human approvals and opt-in assistant --- tests/test_agent.py | 183 ++++++++++++++++++++++++++++++++++++++++++ ui/assistant.html | 88 ++++++++++++++++++++ ui/frame_agent.py | 140 ++++++++++++++++++++++++++++++++ ui/frame_assistant.py | 53 ++++++++++++ ui/frame_mcp.py | 150 ++++++++++++++++++++++++++++++++++ ui/index.html | 1 + ui/server.py | 25 +++++- 7 files changed, 639 insertions(+), 1 deletion(-) create mode 100644 tests/test_agent.py create mode 100644 ui/assistant.html create mode 100644 ui/frame_agent.py create mode 100644 ui/frame_assistant.py create mode 100644 ui/frame_mcp.py diff --git a/tests/test_agent.py b/tests/test_agent.py new file mode 100644 index 0000000..1006583 --- /dev/null +++ b/tests/test_agent.py @@ -0,0 +1,183 @@ +"""MCP protocol, exact-action approvals and explicit assistant data sharing.""" +import io +import json +import os +from pathlib import Path +import subprocess +import sys +import tempfile +import threading +import unittest +from unittest import mock +from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer + +sys.path.insert(0, str(Path(__file__).resolve().parents[1] / 'ui')) +import frame_agent as agent +import frame_assistant as assistant +import frame_mcp as mcp +import server + + +class Approvals(unittest.TestCase): + def test_requires_human_decision_exact_action_and_single_use(self): + gate = agent.Approvals() + action = {'name': 'power', 'arguments': {'action': 'reboot'}} + token = gate.request(action)['confirmation'] + with self.assertRaises(ValueError): + gate.consume(token, action) + gate.decide(token, True) + with self.assertRaises(ValueError): + gate.consume(token, {'name': 'power', 'arguments': {'action': 'poweroff'}}) + gate.consume(token, action) + with self.assertRaises(ValueError): + gate.consume(token, action) + + def test_expiry_rejection_and_non_boolean_approval(self): + gate = agent.Approvals() + token = gate.request({})['confirmation'] + gate.decide(token, 'true') + with self.assertRaises(ValueError): + gate.inspect(token) + token = gate.request({})['confirmation'] + with mock.patch.object(agent.time, 'monotonic', return_value=float('inf')): + with self.assertRaises(ValueError): + gate.decide(token, True) + + def test_concurrent_consumption_executes_once(self): + gate = agent.Approvals() + token = gate.request({})['confirmation'] + gate.decide(token, True) + results = [] + def consume(): + try: + gate.consume(token, {}) + results.append(True) + except ValueError: + results.append(False) + threads = [threading.Thread(target=consume) for _ in range(8)] + for thread in threads: thread.start() + for thread in threads: thread.join() + self.assertEqual(results.count(True), 1) + + def test_action_never_runs_before_approval(self): + with mock.patch.object(agent, 'approvals', agent.Approvals()), mock.patch.object(server, 'flatpak') as install: + body = {'name': 'install', 'arguments': {'id': 'org.example.App'}} + result = agent.call(server, body) + install.assert_not_called() + body['confirmation'] = result['confirmation'] + with self.assertRaises(ValueError): agent.call(server, body) + agent.approvals.decide(body['confirmation'], True) + agent.call(server, body) + install.assert_called_once_with({'id': 'org.example.App', 'action': 'install'}) + with self.assertRaises(ValueError): agent.call(server, body) + + def test_file_content_change_invalidates_approval(self): + with tempfile.TemporaryDirectory() as tmp, mock.patch.object(agent, 'approvals', agent.Approvals()), mock.patch.object(server, 'push_file') as push: + path = Path(tmp) / 'note.txt' + path.write_text('first') + body = {'name': 'send_file', 'arguments': {'path': str(path)}} + result = agent.call(server, body) + agent.approvals.decide(result['confirmation'], True) + body['confirmation'] = result['confirmation'] + path.write_text('second') + with self.assertRaises(ValueError): agent.call(server, body) + push.assert_not_called() + + def test_no_arbitrary_commands_or_arguments(self): + for name, args in [('shell', {'command': 'true'}), ('panel', {'id': 'org.example.App', 'args': '--evil'}), + ('power', {'action': 'factory-reset'}), ('send_text', {'text': ''})]: + with self.assertRaises(ValueError): agent.call(server, {'name': name, 'arguments': args}) + + +class Assistant(unittest.TestCase): + def setUp(self): + self.received = [] + owner = self + class Endpoint(BaseHTTPRequestHandler): + def log_message(self, *args): pass + def do_POST(self): + owner.received.append((dict(self.headers), json.loads(self.rfile.read(int(self.headers['Content-Length']))))) + if self.path == '/redirect': + self.send_response(302) + self.send_header('Location', '/other') + self.end_headers() + return + data = json.dumps({'choices': [{'message': {'content': ''}}]}).encode() + self.send_response(200) + self.send_header('Content-Length', str(len(data))) + self.end_headers() + self.wfile.write(data) + self.httpd = ThreadingHTTPServer(('127.0.0.1', 0), Endpoint) + self.thread = threading.Thread(target=self.httpd.serve_forever, daemon=True) + self.thread.start() + self.body = {'endpoint': 'http://127.0.0.1:%d/chat' % self.httpd.server_port, 'model': 'local', 'prompt': 'Hello', 'consent': True} + + def tearDown(self): + self.httpd.shutdown() + self.httpd.server_close() + self.thread.join() + + def test_no_opt_in_no_request_or_capture(self): + capture = mock.Mock() + for consent in (False, None, 'true', 1): + with self.assertRaises(ValueError): assistant.chat({**self.body, 'consent': consent, 'screenshot': True}, capture) + capture.assert_not_called() + self.assertEqual(self.received, []) + + def test_text_only_keyless_and_optional_screenshot(self): + capture = mock.Mock(return_value=b'png') + self.assertIn('script', assistant.chat(self.body, capture)['reply']) + capture.assert_not_called() + headers, body = self.received[-1] + self.assertNotIn('Authorization', headers) + self.assertEqual(body['messages'], [{'role': 'user', 'content': 'Hello'}]) + assistant.chat({**self.body, 'screenshot': True, 'key': 'test-key'}, capture) + capture.assert_called_once() + headers, body = self.received[-1] + self.assertEqual(headers['Authorization'], 'Bearer test-key') + self.assertEqual(body['messages'][0]['content'][1]['image_url']['url'], 'data:image/png;base64,cG5n') + + def test_redirects_do_not_forward_context_or_credentials(self): + with self.assertRaises(ValueError): + assistant.chat({**self.body, 'endpoint': self.body['endpoint'].replace('/chat', '/redirect'), 'key': 'secret'}, mock.Mock()) + self.assertEqual(len(self.received), 1) + + def test_bad_urls_fail_before_capture(self): + for url in ('file:///etc/passwd', 'http://example.com/chat', 'https://user:pass@example.com', 'https://example.com?key=secret'): + capture = mock.Mock() + with self.assertRaises(ValueError): assistant.chat({**self.body, 'endpoint': url, 'screenshot': True}, capture) + capture.assert_not_called() + + +class Protocol(unittest.TestCase): + def test_stdio_initialize_list_call_errors_and_eof(self): + messages = [ + {'jsonrpc': '2.0', 'id': 1, 'method': 'initialize', 'params': {'protocolVersion': '2025-06-18'}}, + {'jsonrpc': '2.0', 'method': 'notifications/initialized'}, + {'jsonrpc': '2.0', 'id': 2, 'method': 'tools/list'}, + {'jsonrpc': '2.0', 'id': 3, 'method': 'tools/call', 'params': {'name': 'shell'}}, + {'jsonrpc': '2.0', 'id': 4, 'method': 'ping'}, + ] + result = subprocess.run([sys.executable, str(Path(mcp.__file__))], input='\n'.join(map(json.dumps, messages)) + '\n', text=True, capture_output=True, timeout=10) + self.assertEqual(result.returncode, 0, result.stderr) + replies = list(map(json.loads, result.stdout.splitlines())) + self.assertEqual([r['id'] for r in replies], [1, 2, 3, 4]) + self.assertEqual(replies[0]['result']['protocolVersion'], '2025-06-18') + self.assertIn('screenshot', [t['name'] for t in replies[1]['result']['tools']]) + self.assertTrue(replies[2]['result']['isError']) + + def test_mcp_cannot_approve_and_returns_review_url(self): + client = mock.Mock(url='http://127.0.0.1:47810') + client.request.return_value = {'approvalPath': '/assistant#confirm=token'} + result = mcp.call(client, 'power', {'action': 'reboot'}) + self.assertIn('http://127.0.0.1:47810/assistant', result['content'][0]['text']) + with self.assertRaises(ValueError): mcp.call(client, 'approve', {'confirmation': 'token'}) + with self.assertRaises(ValueError): mcp.call(client, 'status', {'path': '/api/open'}) + + def test_loopback_only_backend(self): + for url in ('https://example.com', 'http://127.0.0.1/api', 'http://secret@localhost:1234', 'file:///tmp/x'): + with self.assertRaises(ValueError): mcp.Client(url) + + +if __name__ == '__main__': + unittest.main() diff --git a/ui/assistant.html b/ui/assistant.html new file mode 100644 index 0000000..e0a0770 --- /dev/null +++ b/ui/assistant.html @@ -0,0 +1,88 @@ + + + + +Frame Control · Assistant + +

Frame Control · Assistant

Back to Frame Control
+ +
+

Ask your chosen model

+

Nothing is sent until you opt in and press Send. Each request sends only the message below and, if selected, a fresh headset screenshot. Replies cannot operate your Frame.

+
+
Endpoint and model settings + +Use an OpenAI-compatible endpoint. Loopback means the computer running Frame Control. Remote endpoints require HTTPS. + + +Settings, keys and messages stay in this page’s memory. Reload or close to clear them. No analytics, saved chat history or automatic model discovery.
+ + + + +
+


+

+
+
diff --git a/ui/frame_agent.py b/ui/frame_agent.py
new file mode 100644
index 0000000..ae5a07c
--- /dev/null
+++ b/ui/frame_agent.py
@@ -0,0 +1,140 @@
+"""Agent actions and one-use human approvals. No model SDK or network calls here."""
+import hashlib
+from pathlib import Path
+import secrets
+import shutil
+import subprocess
+import threading
+import time
+
+
+class Approvals:
+    def __init__(self):
+        self.pending = {}
+        self.lock = threading.Lock()
+
+    def request(self, action):
+        with self.lock:
+            now = time.monotonic()
+            self.pending = {k: v for k, v in self.pending.items() if v['expires'] > now}
+            if len(self.pending) >= 100:
+                raise ValueError('Too many pending approvals; wait five minutes')
+            token = secrets.token_urlsafe(24)
+            self.pending[token] = {'action': action, 'approved': False, 'expires': now + 300}
+        return {'confirmation': token, 'action': action, 'approvalPath': '/assistant#confirm=' + token,
+                'message': 'Ask the user to review and approve this action in Frame Control, then retry with confirmation. Expires in five minutes.'}
+
+    def entry(self, token):
+        entry = self.pending.get(token)
+        if not entry or entry['expires'] <= time.monotonic():
+            raise ValueError('Approval expired or unknown; request a new one')
+        return entry
+
+    def inspect(self, token):
+        with self.lock:
+            entry = self.entry(token)
+            return {'action': entry['action'], 'approved': entry['approved']}
+
+    def decide(self, token, accept):
+        with self.lock:
+            entry = self.entry(token)
+            if accept is True:
+                entry['approved'] = True
+            else:
+                del self.pending[token]
+        return {'message': 'Approved for one use' if accept is True else 'Rejected'}
+
+    def consume(self, token, action):
+        with self.lock:
+            entry = self.entry(token)
+            if entry['action'] != action or not entry['approved']:
+                raise ValueError('This exact action needs approval in Frame Control')
+            del self.pending[token]  # consume before starting, including on failure
+
+
+approvals = Approvals()
+
+
+def validate(name, args):
+    fields = {
+        'launch': {'appid'}, 'install': {'id'}, 'uninstall': {'id'},
+        'send_text': {'text'}, 'send_file': {'path'}, 'panel': {'id'},
+        'power': {'action'}, 'keep_awake': {'action'},
+    }
+    if name not in fields or not isinstance(args, dict) or set(args) != fields[name]:
+        raise ValueError('Unknown action or arguments')
+    if any(not isinstance(v, str) or not v or len(v) > 65536 for v in args.values()):
+        raise ValueError('Arguments must be nonempty strings (maximum 65536 characters)')
+    if name == 'power' and args['action'] not in ('suspend', 'reboot', 'poweroff'):
+        raise ValueError('Unknown power action')
+    if name == 'keep_awake' and args['action'] not in ('on', 'off', 'status'):
+        raise ValueError('Expected on, off or status')
+    action = {'name': name, 'arguments': dict(args)}
+    if name == 'send_file':
+        path = Path(args['path']).expanduser().resolve(strict=True)
+        if not path.is_file() or path.stat().st_size > 16 * 1024**2:
+            raise ValueError('Choose a regular file of at most 16 MiB')
+        # Bind approval to bytes, not just a mutable filename.
+        with path.open('rb') as stream:
+            data = stream.read(16 * 1024**2 + 1)
+        if len(data) > 16 * 1024**2:
+            raise ValueError('File grew beyond 16 MiB')
+        action['arguments']['path'] = str(path)
+        action['sha256'] = hashlib.sha256(data).hexdigest()
+        action['bytes'] = len(data)
+    return action
+
+
+def call(server, body):
+    name, args = body.get('name'), body.get('arguments', {})
+    action = validate(name, args)
+    if name in ('install', 'uninstall', 'panel') and not server.FLATPAK_ID.fullmatch(args['id']):
+        raise ValueError('Expected a Flatpak application ID')
+    if name == 'launch' and not server.APPID.fullmatch(args['appid']):
+        raise ValueError('Expected a Steam app ID')
+    if name == 'keep_awake' and args['action'] == 'status':
+        return keep_awake(server, 'status')
+    token = body.get('confirmation')
+    if not token:
+        return approvals.request(action)
+    approvals.consume(token, action)
+    if name == 'launch':
+        return server.launch(args)
+    if name in ('install', 'uninstall'):
+        return server.flatpak({**args, 'action': name})
+    if name == 'send_text':
+        return server.clipboard(args)
+    if name == 'send_file':
+        # Stage the reviewed bytes before the existing transfer helper reads them.
+        import tempfile
+        with tempfile.TemporaryDirectory(prefix='frame-agent-') as tmp:
+            source = Path(action['arguments']['path'])
+            with source.open('rb') as stream:
+                data = stream.read(16 * 1024**2 + 1)
+            if hashlib.sha256(data).hexdigest() != action['sha256']:
+                raise ValueError('File changed after approval')
+            staged = Path(tmp) / source.name
+            staged.write_bytes(data)
+            return {'message': server.push_file(staged)}
+    if name == 'power':
+        if server.LOCAL:
+            raise ValueError('Use the Frame Control power controls to enter the password; MCP never takes passwords')
+        return server.open_thing({'what': args['action']})
+    if name == 'keep_awake':
+        return keep_awake(server, args['action'])
+    return run_script(server, 'panel-on-frame.sh', [args['id']])
+
+
+def run_script(server, name, args):
+    script = server.HERE.parent / 'scripts' / name
+    if not script.exists() or not shutil.which('zsh') or server.LOCAL:
+        raise ValueError(name + ' requires a computer with zsh and the matching script installed')
+    result = subprocess.run(['zsh', str(script), *args], capture_output=True, text=True, timeout=60)
+    if result.returncode:
+        raise ValueError(result.stderr.strip() or 'Script failed')
+    return {'message': result.stdout.strip()}
+
+
+def keep_awake(server, action):
+    # PR #16 owns this interface. Never silently change timers or claim a lease.
+    return run_script(server, 'keep-awake.sh', [action])
diff --git a/ui/frame_assistant.py b/ui/frame_assistant.py
new file mode 100644
index 0000000..c76c1e0
--- /dev/null
+++ b/ui/frame_assistant.py
@@ -0,0 +1,53 @@
+"""Explicit, per-request forwarding to a user-chosen chat-completions endpoint."""
+import base64
+import json
+from urllib.parse import urlsplit
+from urllib.request import HTTPRedirectHandler, ProxyHandler, Request, build_opener
+
+
+class NoRedirect(HTTPRedirectHandler):
+    def redirect_request(self, *args, **kwargs):
+        raise ValueError('Endpoint redirected; enter its final URL explicitly')
+
+
+def chat(body, screenshot):
+    if body.get('consent') is not True:
+        raise ValueError('Opt in before sending a message')
+    endpoint, model, prompt = (body.get(k) for k in ('endpoint', 'model', 'prompt'))
+    if any(not isinstance(v, str) or not v.strip() for v in (endpoint, model, prompt)):
+        raise ValueError('Endpoint, model and message are required')
+    if len(prompt) > 32000 or len(model) > 200 or len(endpoint) > 2048:
+        raise ValueError('Message, model or endpoint is too long')
+    url = urlsplit(endpoint)
+    if not url.hostname or url.username or url.password or url.fragment or url.query:
+        raise ValueError('Use an endpoint URL without credentials, query or fragment')
+    if url.scheme != 'https' and not (url.scheme == 'http' and url.hostname in ('localhost', '127.0.0.1', '::1')):
+        raise ValueError('Use HTTPS, or HTTP on loopback for a local model')
+    key = body.get('key', '')
+    if not isinstance(key, str) or len(key) > 4096 or '\n' in key or '\r' in key:
+        raise ValueError('Invalid API key')
+    content = prompt
+    if body.get('screenshot') is True:
+        png = screenshot()
+        if len(png) > 12 * 1024**2:
+            raise ValueError('Screenshot is too large')
+        content = [{'type': 'text', 'text': prompt}, {'type': 'image_url', 'image_url': {
+            'url': 'data:image/png;base64,' + base64.b64encode(png).decode()}}]
+    payload = {'model': model, 'messages': [{'role': 'user', 'content': content}], 'stream': False}
+    headers = {'Content-Type': 'application/json'}
+    if key:
+        headers['Authorization'] = 'Bearer ' + key
+    request = Request(endpoint, data=json.dumps(payload).encode(), headers=headers)
+    # No environment proxy or redirects: credentials/context go only to the chosen URL.
+    try:
+        with build_opener(ProxyHandler({}), NoRedirect()).open(request, timeout=60) as response:
+            raw = response.read(2 * 1024**2 + 1)
+        if len(raw) > 2 * 1024**2:
+            raise ValueError('Endpoint response is too large')
+        answer = json.loads(raw)['choices'][0]['message']['content']
+        if not isinstance(answer, str):
+            raise ValueError('Expected a text reply')
+    except Exception:
+        # Provider error bodies and URLs can contain credentials or echoed prompts.
+        raise ValueError('Endpoint request failed or returned an unsupported reply; check URL, model and credentials') from None
+    return {'reply': answer}
diff --git a/ui/frame_mcp.py b/ui/frame_mcp.py
new file mode 100644
index 0000000..6719dcf
--- /dev/null
+++ b/ui/frame_mcp.py
@@ -0,0 +1,150 @@
+#!/usr/bin/env python3
+"""Key-free stdio MCP adapter for an already running Frame Control HTTP server."""
+import argparse
+import base64
+import json
+import os
+import sys
+from urllib.parse import urlencode, urlsplit
+from urllib.error import HTTPError
+from urllib.request import ProxyHandler, Request, build_opener, HTTPRedirectHandler
+
+MAX_LINE = 1024 * 1024
+
+
+class NoRedirect(HTTPRedirectHandler):
+    def redirect_request(self, *args, **kwargs):
+        raise ValueError('Frame Control must not redirect')
+
+
+class Client:
+    def __init__(self, url, key='1'):
+        parsed = urlsplit(url)
+        if parsed.scheme != 'http' or parsed.hostname not in ('localhost', '127.0.0.1') or parsed.path not in ('', '/') or parsed.query or parsed.fragment or parsed.username or parsed.password:
+            raise ValueError('Frame Control URL must be HTTP loopback with no path or credentials')
+        self.url, self.key = url.rstrip('/'), key
+        self.opener = build_opener(ProxyHandler({}), NoRedirect())
+
+    def request(self, path, body=None, image=False):
+        req = Request(self.url + path, data=None if body is None else json.dumps(body).encode(),
+                      headers={'X-Frame-UI': self.key, 'Content-Type': 'application/json'})
+        try:
+            with self.opener.open(req, timeout=360) as res:
+                data = res.read(16 * 1024**2 + 1)
+        except HTTPError as exc:
+            with exc:
+                raw = exc.read(65536)
+            try:
+                message = json.loads(raw).get('error', 'HTTP ' + str(exc.code))
+            except (ValueError, AttributeError):
+                message = 'HTTP ' + str(exc.code)
+            raise ValueError(str(message)) from None
+        if len(data) > 16 * 1024**2:
+            raise ValueError('Frame Control response too large')
+        return data if image else json.loads(data)
+
+
+def tool(name, description, properties=None, required=None, read=False):
+    return {'name': name, 'description': description, 'inputSchema': {
+        'type': 'object', 'properties': properties or {}, 'required': required or [], 'additionalProperties': False},
+        'annotations': {'readOnlyHint': read, 'destructiveHint': not read, 'openWorldHint': True}}
+
+
+def string(description):
+    return {'type': 'string', 'description': description}
+
+
+TOOLS = [tool('status', 'Read battery, services and installed apps.', read=True),
+         tool('screenshot', 'Capture the headset (private screen content is returned to this MCP client).',
+              {'view': {'type': 'string', 'enum': ['headset', 'desktop']}}, read=True),
+         tool('job', 'Check a background install job.', {'id': string('Job ID')}, ['id'], read=True)]
+for name, field, description in [
+    ('launch', 'appid', 'Launch an installed Steam app by ID.'),
+    ('install', 'id', 'Install a free Flatpak from Flathub to the user account.'),
+    ('uninstall', 'id', 'Uninstall a user Flatpak.'),
+    ('send_text', 'text', 'Send text to the Frame desktop clipboard.'),
+    ('send_file', 'path', 'Send a file (up to 16 MiB) from the HTTP server computer to Frame Downloads.'),
+    ('panel', 'id', 'Open an installed Flatpak as a floating panel; needs zsh on the computer.'),
+    ('power', 'action', 'suspend, reboot or poweroff. Opens a terminal for the user password.'),
+    ('keep_awake', 'action', 'on, off or status using the optional PR #16 script. on changes idle timers; off restores them. Never automatic.'),
+]:
+    TOOLS.append(tool(name, description + ' Mutations require user approval at the returned approvalUrl; retry with its confirmation token. Never approve on the user’s behalf.',
+                      {field: string(description), 'confirmation': string('Token returned by a previous call, after the user approves')}, [field]))
+
+
+def call(client, name, args):
+    spec = next((t for t in TOOLS if t['name'] == name), None)
+    if not spec or not isinstance(args, dict):
+        raise ValueError('Unknown tool or invalid arguments')
+    schema = spec['inputSchema']
+    if set(args) - set(schema['properties']) or set(schema['required']) - set(args):
+        raise ValueError('Unknown or missing arguments')
+    if any(not isinstance(v, str) for v in args.values()):
+        raise ValueError('Arguments must be strings')
+    if name == 'screenshot':
+        view = args.get('view', 'headset')
+        if view not in ('headset', 'desktop'):
+            raise ValueError('Unknown screenshot view')
+        png = client.request('/api/screenshot?' + urlencode({'view': view}), image=True)
+        return {'content': [{'type': 'image', 'mimeType': 'image/png', 'data': base64.b64encode(png).decode()}]}
+    if name in ('status', 'job'):
+        result = client.request('/api/' + name + ('?' + urlencode(args) if args else ''))
+    else:
+        args = dict(args)
+        confirmation = args.pop('confirmation', None)
+        result = client.request('/api/agent/call', {'name': name, 'arguments': args, 'confirmation': confirmation})
+        if 'approvalPath' in result:
+            result['approvalUrl'] = client.url + result['approvalPath']
+    return {'content': [{'type': 'text', 'text': json.dumps(result)}]}
+
+
+def dispatch(client, message):
+    if not isinstance(message, dict) or message.get('jsonrpc') != '2.0' or not isinstance(message.get('method'), str):
+        return {'jsonrpc': '2.0', 'id': None, 'error': {'code': -32600, 'message': 'Invalid request'}}
+    if 'id' not in message:
+        return None
+    method, params = message['method'], message.get('params', {})
+    response = {'jsonrpc': '2.0', 'id': message['id']}
+    if not isinstance(params, dict):
+        return {**response, 'error': {'code': -32602, 'message': 'Invalid params'}}
+    if method == 'initialize':
+        requested = params.get('protocolVersion')
+        result = {'protocolVersion': requested if requested in ('2024-11-05', '2025-03-26', '2025-06-18') else '2025-06-18',
+                  'capabilities': {'tools': {}}, 'serverInfo': {'name': 'frame-control', 'version': '1.0.0'}}
+    elif method == 'ping':
+        result = {}
+    elif method == 'tools/list':
+        result = {'tools': TOOLS}
+    elif method == 'tools/call':
+        try:
+            result = call(client, params.get('name'), params.get('arguments', {}))
+        except Exception as exc:
+            result = {'isError': True, 'content': [{'type': 'text', 'text': 'Frame Control: ' + str(exc)}]}
+    else:
+        return {**response, 'error': {'code': -32601, 'message': 'Method not found'}}
+    return {**response, 'result': result}
+
+
+def main():
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument('--url', default='http://127.0.0.1:47810')
+    args = parser.parse_args()
+    client = Client(args.url, os.environ.get('FRAME_UI_KEY', '1'))
+    while True:
+        line = sys.stdin.buffer.readline(MAX_LINE + 1)
+        if not line:
+            break
+        if len(line) > MAX_LINE:
+            print('MCP request too large', file=sys.stderr)
+            return 1
+        try:
+            response = dispatch(client, json.loads(line))
+        except (ValueError, UnicodeError):
+            response = {'jsonrpc': '2.0', 'id': None, 'error': {'code': -32700, 'message': 'Parse error'}}
+        if response is not None:
+            print(json.dumps(response), flush=True)
+    return 0
+
+
+if __name__ == '__main__':
+    sys.exit(main())
diff --git a/ui/index.html b/ui/index.html
index 8ac1265..3b106a8 100644
--- a/ui/index.html
+++ b/ui/index.html
@@ -591,6 +591,7 @@
   
 
   

+  
Assistant and AI agents
Use your own model endpoint, or review a proposed MCP action. Nothing is sent to a model until you opt in.
Open assistant

   

     

       
Send to Frame

diff --git a/ui/server.py b/ui/server.py
index dd86ff6..d5ca4d1 100755
--- a/ui/server.py
+++ b/ui/server.py
@@ -36,6 +36,8 @@ from urllib.parse import parse_qs, unquote, urlparse
 # sys.path, so add it for the sibling modules below.
 sys.path.insert(0, str(Path(__file__).resolve().parent))
 
+import frame_agent  # noqa: E402
+import frame_assistant  # noqa: E402
 import frame_android  # noqa: E402
 import frame_apk_versions  # noqa: E402
 import frame_catalog  # noqa: E402
@@ -1227,7 +1229,20 @@ def _sweep_one(prefix, d):
         pass
 
 
-POST = {"/api/android/display": android_display, "/api/android": android, "/api/titles": titles, "/api/launch": launch, "/api/steam": steam, "/api/volume": set_volume, "/api/clipboard": clipboard,
+def agent_call(body):
+    return frame_agent.call(sys.modules[__name__], body)
+
+
+def assistant_chat(body):
+    return frame_assistant.chat(body, headset_view)
+
+
+def agent_approval(body):
+    return frame_agent.approvals.decide(body.get("confirmation"), body.get("accept"))
+
+
+POST = {"/api/agent/call": agent_call, "/api/agent/approval": agent_approval,
+        "/api/assistant/chat": assistant_chat, "/api/android/display": android_display, "/api/android": android, "/api/titles": titles, "/api/launch": launch, "/api/steam": steam, "/api/volume": set_volume, "/api/clipboard": clipboard,
         "/api/flatpak": flatpak, "/api/open": open_thing, "/api/shots/save": save_shots,
         "/api/webinstall/check": webinstall_check, "/api/webinstall/start": webinstall_start,
         "/api/webinstall/cancel": webinstall_cancel}
@@ -1331,6 +1346,12 @@ class Handler(BaseHTTPRequestHandler):
         try:
             if path in ("/", "/index.html"):
                 self.send_bytes((HERE / "index.html").read_bytes(), "text/html; charset=utf-8")
+            elif path == "/assistant":
+                page = (HERE / "assistant.html").read_text().replace("__FRAME_KEY__", json.dumps(UI_KEY).replace("<", "\\u003c"))
+                self.send_bytes(page.encode(), "text/html; charset=utf-8")
+            elif path == "/api/agent/approval":
+                token = (parse_qs(url.query).get("confirmation") or [""])[0]
+                self.send_json(frame_agent.approvals.inspect(token))
             elif path == "/api/host":
                 self.send_json({"os": "SteamOS", "fileManager": None, "computer": DEVICE, "mobile": True} if LOCAL else
                                {"os": frame_host.NAME, "fileManager": frame_host.FILE_MANAGER,
@@ -1377,6 +1398,8 @@ class Handler(BaseHTTPRequestHandler):
                 self.send_json({"error": "not found"}, 404)
         except Failure as e:
             self.send_error_json(str(e), e.status, e.apk)
+        except ValueError as e:
+            self.send_json({"error": str(e)}, 400)
         except frame_android.FrameError as e:
             self.send_error_json(str(e), 502)
         except Exception as e: