Setting the file. One moment.
Run Instance AI · N8n:testmax N8n AI · n8n-io/n8n · Skills Docs
Back to the top of the page — line 158
This file
Number 40.17
Position 17 of 24
Type Python
Size 12 KB
Lines 329 scripts/ run_instance_ai.py
Python · 329 lines · 12 KB
16 """
17
18 import argparse
19 import json
20 import os
21 import queue
22 import re
23 import shutil
24 import subprocess
25 import sys
26 import threading
27 import time
28 import uuid
29 from datetime import datetime, timezone
30 from pathlib import Path
31
32 from live import enabled as live_enabled
33 from live import open_url, recording
34 from n8n_api import N8nClient, setting
35
36 DEFAULT_TEXT_ANSWER = 'Use sensible defaults and continue.'
37 HERE = Path( __file__ ).resolve().parent
38
39
40 class Recorder :
41 """Video of the thread in the n8n UI, made by record_iai.mjs."""
42
43 def __init__ (self, out):
44 self .proc, self .lines = None , queue.Queue()
45 node = shutil.which( 'node' )
46 if not recording() or not node:
47 return
48 live_flag = [ '--live' ] if live_enabled() else []
49 self .proc = subprocess.Popen([node, str ( HERE / 'record_iai.mjs' ), str (out.resolve()), * live_flag], cwd = HERE ,
50 stdin = subprocess. PIPE , stdout = subprocess. PIPE ,
51 stderr = (out / 'record.log' ).open( 'w' ), text = True )
52 threading.Thread( target = self ._read, daemon = True ).start()
53
54 def _read (self):
55 for line in self .proc.stdout:
56 self .lines.put(line.strip())
57
58 def wait_for (self, word, timeout):
59 deadline = time.time() + timeout
60 while self .proc and time.time() < deadline:
61 try :
62 if self .lines.get( timeout = min ( max (deadline - time.time(), 0.1 ), 1 )) == word:
63 return True
64 except queue.Empty:
65 if self .proc.poll() is not None :
66 return False
67 return False
68
69 def send (self, line):
70 if self .proc and self .proc.poll() is None :
71 self .proc.stdin.write(line + ' \n ' )
72 self .proc.stdin.flush()
73
74 def stop (self):
75 if not self .proc:
76 return
77 self .send( 'stop' )
78 self .proc.stdin.close()
79 try :
80 self .proc.wait( timeout = 120 )
81 except subprocess.TimeoutExpired:
82 self .proc.kill()
83
84
85 ANTHROPIC_EFFORTS = ( 'low' , 'medium' , 'high' , 'xhigh' , 'max' )
86 CUSTOM_EFFORTS = ( 'none' , 'minimal' , * ANTHROPIC_EFFORTS )
87 # Keep in sync with instance-ai.config.ts and CUSTOM_MODEL_DEFAULTS in utils/custom-model-defaults.ts.
88 DEFAULT_INSTANCE_AI_MODEL = 'anthropic/claude-opus-5-5'
89 CUSTOM_MODEL_EFFORTS = (( 'kimi-k3' , 'low' ), ( 'glm-5.2' , 'medium' ))
90
91
92 def configured_effort ():
93 """The orchestrator effort that n8n uses for N8N_INSTANCE_AI_MODEL, with the rules of
94 packages/@n8n/instance-ai/src/agent/apply-agent-thinking.ts. n8n does not report the effort, so this
95 reads the settings that the server reads. None when the provider has no fixed effort."""
96 model = (setting( 'N8N_INSTANCE_AI_MODEL' ) or DEFAULT_INSTANCE_AI_MODEL ).lower()
97 provider = model.split( '/' , 1 )[ 0 ] if '/' in model else ''
98 effort = (setting( 'N8N_INSTANCE_AI_REASONING_EFFORT' ) or '' ).strip().lower()
99 if provider in ( 'anthropic' , 'google-vertex-anthropic' ):
100 return effort if effort in ANTHROPIC_EFFORTS else 'medium'
101 if provider in ( 'custom' , 'moonshotai' ):
102 if effort in CUSTOM_EFFORTS :
103 return effort
104 return next ((e for match, e in CUSTOM_MODEL_EFFORTS if match in model), None )
105 if provider == 'openai' :
106 return 'medium' if 'gpt-5.6' in model else 'high'
107 if provider == 'xai' and 'grok-4.5' in model:
108 return 'medium'
109 return None
110
111
112 def now_iso ():
113 return datetime.now(timezone.utc).isoformat()
114
115
116 def local_time_zone ():
117 zone = os.environ.get( 'TZ' )
118 if zone:
119 return zone
120 link = os.path.realpath( '/etc/localtime' )
121 return link.split( 'zoneinfo/' , 1 )[ 1 ] if 'zoneinfo/' in link else 'UTC'
122
123
124 def answer_questions (questions):
125 answers = []
126 for q in questions or []:
127 options = q.get( 'options' ) or []
128 if q.get( 'type' ) in ( 'single' , 'multi' ) and options:
129 answers.append({ 'questionId' : q[ 'id' ], 'selectedOptions' : [options[ 0 ]]})
130 elif q.get( 'required' ):
131 answers.append({ 'questionId' : q[ 'id' ], 'selectedOptions' : [], 'customText' : DEFAULT_TEXT_ANSWER })
132 else :
133 answers.append({ 'questionId' : q[ 'id' ], 'selectedOptions' : [], 'skipped' : True })
134 return { 'kind' : 'questions' , 'answers' : answers}
135
136
137 def decide (payload):
138 """Return (confirm body, short label) for one confirmation-request payload."""
139 input_type = payload.get( 'inputType' ) or 'approval'
140 if payload.get( 'credentialDestination' ):
141 origin = payload[ 'credentialDestination' ].get( 'origin' )
142 return { 'kind' : 'credentialDestination' , 'origin' : origin, 'approved' : False }, 'defer credential destination'
143 if payload.get( 'setupRequests' ) or payload.get( 'credentialRequests' ) or payload.get( 'credentialFlow' ):
144 return { 'kind' : 'approval' , 'approved' : False }, 'defer setup'
145 if payload.get( 'domainAccess' ):
146 return { 'kind' : 'domainAccessApprove' , 'domainAccessAction' : 'allow_once' }, 'allow domain once'
147 if payload.get( 'mcpConnectRequest' ):
148 return { 'kind' : 'mcpConnect' , 'approved' : False }, 'deny MCP connect'
149 if input_type == 'resource-decision' :
150 return { 'kind' : 'resourceDecision' , 'resourceDecision' : 'denyOnce' }, 'deny local resource'
151 if input_type == 'questions' :
152 return answer_questions(payload.get( 'questions' )), 'answer questions'
153 if input_type == 'text' :
154 return { 'kind' : 'approval' , 'approved' : True , 'userInput' : DEFAULT_TEXT_ANSWER }, 'answer text'
155 return { 'kind' : 'approval' , 'approved' : True }, f 'approve { input_type } '
156
157
158 def read_stream (client, thread_id, out_queue, stop):
159 try :
160 res = client.open_stream( f '/rest/instance-ai/events/ { thread_id } ' )
161 except Exception as e:
162 out_queue.put(( 'error' , str (e)))
163 return
164 out_queue.put(( 'open' , None ))
165 data_lines, event_name = [], None
166 while not stop.is_set():
167 raw = res.readline()
168 if not raw:
169 out_queue.put(( 'closed' , None ))
170 return
171 line = raw.decode( errors = 'replace' ).rstrip( ' \r\n ' )
172 if line == '' :
173 if data_lines:
174 try :
175 out_queue.put(( 'event' , (event_name, json.loads( ' \n ' .join(data_lines)))))
176 except json.JSONDecodeError:
177 pass
178 data_lines, event_name = [], None
179 elif line.startswith( 'data:' ):
180 data_lines.append(line[ 5 :].lstrip())
181 elif line.startswith( 'event:' ):
182 event_name = line[ 6 :].strip()
183
184
185 def collect_workflow_ids (value, found):
186 text = json.dumps(value)
187 for match in re.finditer( r '"workflowId" \s * : \s * " ([ A-Za-z0-9 ] {10,} ) "' , text):
188 if match.group( 1 ) not in found:
189 found.append(match.group( 1 ))
190
191
192 def main ():
193 ap = argparse.ArgumentParser()
194 ap.add_argument( '--out' , required = True )
195 ap.add_argument( '--prompt-file' , required = True )
196 ap.add_argument( '--timeout-min' , type = float , default = 30 )
197 args = ap.parse_args()
198
199 out = Path(args.out)
200 out.mkdir( parents = True , exist_ok = True )
201 prompt = Path(args.prompt_file).read_text().strip()
202
203 # The recorder signs in while the event stream opens, so the video starts soon after the prompt.
204 recorder = Recorder(out)
205 client = N8nClient().login()
206 thread_id = str (uuid.uuid4())
207 events_file = (out / 'events.jsonl' ).open( 'w' )
208 q, stop = queue.Queue(), threading.Event()
209 threading.Thread( target = read_stream, args = (client, thread_id, q, stop), daemon = True ).start()
210
211 kind, detail = q.get( timeout = 30 )
212 if kind != 'open' :
213 sys.exit( f 'Event stream did not open: { detail } ' )
214
215 project = client.request( 'GET' , '/rest/projects/personal' )
216 client.request( 'POST' , '/rest/instance-ai/threads' ,
217 { 'threadId' : thread_id, 'projectId' : project[ 'id' ], 'source' : 'assistant_page' })
218
219 if recorder.proc and not recorder.wait_for( 'ready' , 60 ):
220 print ( f 'recording failed to start, see { out / "record.log" } ' , flush = True )
221 recorder.proc.kill()
222 recorder.proc = None
223
224 started_at = now_iso()
225 chat = client.request( 'POST' , f '/rest/instance-ai/chat/ { thread_id } ' ,
226 { 'message' : prompt, 'timeZone' : local_time_zone()})
227 print ( f 'thread { thread_id } run { chat.get( "runId" ) if isinstance (chat, dict ) else chat } ' , flush = True )
228 thread_url = f ' { client.base_url } /instance-ai/ { thread_id } '
229 # The UI shows the user message only after the chat request, so the recorder opens the thread now.
230 recorder.send( f 'open { thread_url } ' )
231 print ( f 'watch: { thread_url } ' , flush = True )
232 open_url(thread_url)
233
234 workflow_ids, decisions, langsmith_run_ids = [], [], []
235 # The idle wait after run-finish is the runner's, so the run ends at the last run-finish.
236 saw_finish, idle_since, finish_ts = False , None , None
237 deadline = time.time() + args.timeout_min * 60
238 while time.time() < deadline:
239 try :
240 kind, detail = q.get( timeout = 5 )
241 except queue.Empty:
242 kind, detail = 'tick' , None
243
244 if kind == 'event' :
245 _, event = detail
246 events_file.write(json.dumps(event) + ' \n ' )
247 events_file.flush()
248 etype, payload = event.get( 'type' ), event.get( 'payload' ) or {}
249 if etype == 'run-start' :
250 if payload.get( 'langsmithRunId' ):
251 langsmith_run_ids.append(payload[ 'langsmithRunId' ])
252 idle_since, finish_ts = None , None
253 elif etype == 'run-finish' :
254 saw_finish = True
255 finish_ts = event.get( 'ts' ) or time.time() * 1000
256 print ( f 'run-finish { payload.get( "status" , "" ) } ' , flush = True )
257 elif etype == 'tool-call' :
258 print ( f 'tool { payload.get( "toolName" ) } ' , flush = True )
259 elif etype in ( 'tool-result' , 'confirmation-request' ):
260 collect_workflow_ids(payload, workflow_ids)
261 if etype == 'confirmation-request' :
262 body, label = decide(payload)
263 try :
264 client.request( 'POST' , f '/rest/instance-ai/confirm/ { payload[ "requestId" ] } ' , body)
265 ok = True
266 except RuntimeError as e:
267 ok = False
268 label += f ' (failed: { e } )'
269 decisions.append({ 'at' : now_iso(), 'toolName' : payload.get( 'toolName' ),
270 'message' : payload.get( 'message' ), 'decision' : label, 'ok' : ok})
271 print ( f 'confirm: { label } [ { payload.get( "toolName" ) } ]' , flush = True )
272 elif kind in ( 'closed' , 'error' ):
273 print ( f 'event stream { kind } : { detail or "" } ' , flush = True )
274
275 if saw_finish:
276 status = client.request( 'GET' , f '/rest/instance-ai/threads/ { thread_id } /status' )
277 running = [t for t in status.get( 'backgroundTasks' , []) if t.get( 'status' ) == 'running' ]
278 if status.get( 'hasActiveRun' ) or status.get( 'isSuspended' ) or running:
279 idle_since = None
280 elif idle_since is None :
281 idle_since = time.time()
282 elif time.time() - idle_since > 10 :
283 break
284 else :
285 print ( 'timeout: the thread did not go idle' , flush = True )
286
287 finished_at = (datetime.fromtimestamp(finish_ts / 1000 , timezone.utc).isoformat() if finish_ts
288 else now_iso())
289 # Background tasks can still call the LLM after run-finish, so the usage window ends at idle.
290 usage_finished_at = now_iso()
291 stop.set()
292 events_file.close()
293 recorder.stop()
294
295 trace = {}
296 for key, path in (
297 ( 'rich' , f '/threads/ { thread_id } /messages?limit=200' ),
298 ( 'raw' , f '/threads/ { thread_id } /messages?limit=200&raw=true' ),
299 ( 'thread' , f '/threads/ { thread_id } ' ),
300 ( 'debug' , f '/debug/threads/ { thread_id } /runs' ),
301 ):
302 try :
303 trace[key] = client.request( 'GET' , '/rest/instance-ai' + path)
304 except RuntimeError as e:
305 trace[key] = None
306 print ( f 'export { key } failed: { str (e)[: 200 ] } ' , flush = True )
307 (out / 'trace.json' ).write_text(json.dumps(trace, indent = 1 ))
308
309 meta = {
310 'kind' : 'instance-ai' ,
311 'threadId' : thread_id,
312 'startedAt' : started_at,
313 'finishedAt' : finished_at,
314 'usageFinishedAt' : usage_finished_at,
315 'workflowIds' : workflow_ids,
316 'workflowId' : workflow_ids[ - 1 ] if workflow_ids else None ,
317 'langsmithRunIds' : langsmith_run_ids,
318 'decisions' : decisions,
319 'effort' : configured_effort(),
320 'effortSource' : 'n8n settings (N8N_INSTANCE_AI_MODEL, N8N_INSTANCE_AI_REASONING_EFFORT)' ,
321 'threadUrl' : thread_url,
322 'screen' : 'screen.webm' if (out / 'screen.webm' ).exists() else None ,
323 }
324 (out / 'run.json' ).write_text(json.dumps(meta, indent = 2 ))
325 print (json.dumps({k: meta[k] for k in ( 'threadId' , 'workflowId' , 'startedAt' , 'finishedAt' )}, indent = 2 ))
326
327
328 if __name__ == '__main__' :
329 main()