-
Notifications
You must be signed in to change notification settings - Fork 0
Expand file tree
/
Copy path_core.py
More file actions
235 lines (192 loc) · 10.5 KB
/
Copy path_core.py
File metadata and controls
235 lines (192 loc) · 10.5 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
"""DevTools: main orchestrator — register objects, start inspection server."""
from __future__ import annotations
import logging
import os
import sys
from collections.abc import Callable
from typing import TYPE_CHECKING
if TYPE_CHECKING:
from python_devtools._server import _Server
log = logging.getLogger('python-devtools')
def _default_app_id() -> str:
app_id = os.environ.get('DEVTOOLS_APP_ID') or os.environ.get('_DEVTOOLS_APP_ID')
if app_id:
return app_id
script_name = os.path.basename(sys.argv[0] or 'python')
stem, _ = os.path.splitext(script_name)
base = stem or 'python'
return f'{base}-{os.getpid()}'
class DevTools:
"""
Runtime inspection server for Python apps.
Register named objects, start the TCP server, and LLM agents can
connect via the MCP CLI bridge to query live app state.
Threading safety:
By default, resolve/eval calls run inline on the TCP handler thread.
For apps with a main-thread constraint (GUI frameworks, game loops),
call set_main_thread_invoker() with a callback that schedules work
back onto the main thread and returns the result.
LOCAL_TRUSTED mode:
The server binds to loopback only and rejects non-loopback peers.
eval/exec is intentionally unrestricted — this is a dev tool, not
a production service. The warning banner on start makes this explicit.
"""
def __init__(self):
self._namespaces: dict[str, object] = {}
self._server: _Server | None = None
self._invoke_fn: Callable | None = None
self._screenshot_fn: Callable[[], bytes] | None = None
self._winshot_fn: Callable[[str], bytes] | None = None
# ────────────────────────────────────────────────────────────────────
# Registration
# ────────────────────────────────────────────────────────────────────
def register(self, name: str, obj: object) -> None:
"""Register an object under a name for inspection."""
self._namespaces[name] = obj
log.debug(f'devtools: registered {name!r} ({type(obj).__name__})')
def unregister(self, name: str) -> None:
"""Remove a registered object."""
self._namespaces.pop(name, None)
# ────────────────────────────────────────────────────────────────────
# Threading safety
# ────────────────────────────────────────────────────────────────────
def set_main_thread_invoker(self, callback: Callable | None) -> None:
"""
Set a callback that runs resolve/eval on the app's main thread.
The callback signature: callback(fn) -> result
It receives a zero-arg callable, must execute it on the main thread,
and return the result. If None, calls run inline on the TCP thread.
Example (imgui app with frame-synced queue):
def invoke_on_main(fn):
future = concurrent.futures.Future()
main_queue.put((fn, future))
return future.result(timeout=10)
devtools.set_main_thread_invoker(invoke_on_main)
"""
self._invoke_fn = callback
# Propagate to live server if already running
if self._server is not None:
self._server._invoke_fn = callback
def set_screenshot_fn(self, callback: Callable[[], bytes] | None) -> None:
"""
Register a callback that captures the app's current visual state as PNG bytes.
The callback must return PNG-encoded bytes. It will be invoked on the main
thread (via invoke_fn) so it has access to the framebuffer/GL context.
Without this, the screenshot MCP tool returns an error explaining the
capability isn't available.
Example (OpenGL app):
def capture():
w, h = glfw.get_framebuffer_size(window)
pixels = gl.glReadPixels(0, 0, w, h, gl.GL_RGB, gl.GL_UNSIGNED_BYTE)
# ... flip, encode PNG, return bytes
devtools.set_screenshot_fn(capture)
"""
self._screenshot_fn = callback
if self._server is not None:
self._server._screenshot_fn = callback
def set_winshot_fn(self, callback: Callable[[str], bytes] | None) -> None:
"""
Register a callback that renders code in an offscreen window and returns PNG bytes.
The callback signature: (code: str) -> bytes
It receives a code string, sets up an isolated offscreen rendering context,
executes the code within it (e.g., imgui widget calls), captures the rendered
output, and returns PNG-encoded bytes.
This enables the ``winshot`` MCP tool, which is the focused complement to
``screenshot``: where screenshot captures the entire live app, winshot renders
*only* the code the agent passes — a single panel, a test widget, a specific
component in isolation. Not all apps can support this (it requires offscreen
rendering), so the tool gracefully errors if no callback is registered.
Example (imgui app):
def render_winshot(code: str) -> bytes:
# Create FBO, start imgui frame, exec(code), render, readback PNG
...
devtools.set_winshot_fn(render_winshot)
"""
self._winshot_fn = callback
if self._server is not None:
self._server._winshot_fn = callback
# ────────────────────────────────────────────────────────────────────
# Server lifecycle
# ────────────────────────────────────────────────────────────────────
def start(
self,
*,
port: int = 0,
host: str = 'localhost',
readonly: bool = False,
app_id: str | None = None,
) -> None:
"""Start the inspection server in a background thread."""
if self._server is not None:
log.warning('devtools: server already running')
return
resolved_app_id = app_id or _default_app_id()
from python_devtools._server import start_server
self._server = start_server(
self._namespaces,
host=host,
port=port,
app_id=resolved_app_id,
invoke_fn=self._invoke_fn,
readonly=readonly,
)
# Propagate callbacks if already set before start()
if self._screenshot_fn is not None:
self._server._screenshot_fn = self._screenshot_fn
if self._winshot_fn is not None:
self._server._winshot_fn = self._winshot_fn
# LOCAL_TRUSTED banner — make the security posture explicit
log.warning('python-devtools: LOCAL_TRUSTED mode — eval/exec enabled, loopback-only, no auth')
if readonly:
log.warning('python-devtools: readonly mode — mutation tools disabled')
log.info(f'devtools: listening on {self._server.host}:{self._server.port} (app_id={self._server.app_id})')
def stop(self) -> None:
"""Stop the inspection server."""
if self._server is not None:
self._server.shutdown()
self._server = None
# ────────────────────────────────────────────────────────────────────
# Observable state (for GUI indicators)
# ────────────────────────────────────────────────────────────────────
@property
def running(self) -> bool:
return self._server is not None
@property
def readonly(self) -> bool:
return self._server._readonly if self._server else False
@property
def n_clients(self) -> int:
return self._server.n_clients if self._server else 0
@property
def n_commands(self) -> int:
return self._server.n_commands if self._server else 0
@property
def last_command_time(self) -> float:
return self._server.last_command_time if self._server else 0.0
@property
def app_id(self) -> str | None:
return self._server.app_id if self._server else None
# ────────────────────────────────────────────────────────────────────
# Argparse integration
# ────────────────────────────────────────────────────────────────────
def add_arguments(self, parser) -> None:
"""Add --devtools, --devtools-port, --devtools-app-id, --devtools-readonly."""
group = parser.add_argument_group('DevTools')
group.add_argument('--devtools', action='store_true', help='Enable runtime devtools server')
group.add_argument(
'--devtools-port',
type=int,
default=0,
help='DevTools port (default: 0 for automatic free port)',
)
group.add_argument('--devtools-app-id', type=str, default=None, help='Stable app id used by MCP tools')
group.add_argument('--devtools-readonly', action='store_true', help='Disable eval/call/set (read-only mode)')
def from_args(self, args, **namespaces) -> None:
"""Register namespaces and conditionally start from parsed args."""
for name, obj in namespaces.items():
self.register(name, obj)
if getattr(args, 'devtools', False):
port = getattr(args, 'devtools_port', 0)
readonly = getattr(args, 'devtools_readonly', False)
app_id = getattr(args, 'devtools_app_id', None)
self.start(port=port, readonly=readonly, app_id=app_id)