Setting the file. One moment.
Request Recorder · N8n:testmax N8n AI · n8n-io/n8n · Skills Docs
Back to the top of the page
40.15 scripts/ request_recorder.py
Python · 192 lines · 7 KB
15
import
json
16 import threading
17 import time
18 import zlib
19 from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
20 from urllib.parse import urlparse
21
22 # Headers that apply to one connection only. The recorder sets its own values.
23 HOP_HEADERS = { 'connection' , 'keep-alive' , 'proxy-connection' , 'proxy-authorization' , 'te' , 'trailer' ,
24 'transfer-encoding' , 'upgrade' , 'host' , 'content-length' , 'accept-encoding' }
25 CHANGING_KEYS = ( 'tools' , 'instructions' )
26
27
28 def decode_body (data, encoding):
29 encoding = (encoding or '' ).lower()
30 if encoding == 'gzip' :
31 return gzip.decompress(data)
32 if encoding == 'deflate' :
33 return zlib.decompress(data)
34 if encoding == 'zstd' :
35 from compression import zstd
36 return zstd.decompress(data)
37 return data
38
39
40 class RequestRecorder :
41 """Start with start(). Point the CLI at base_url(host). Call stop() when the CLI is done."""
42
43 def __init__ (self, upstream, out_file, bind = '127.0.0.1' ):
44 self .upstream = urlparse(upstream)
45 self .out_file = out_file
46 self .lock = threading.Lock()
47 self .idle = threading.Condition( self .lock)
48 self .active = 0
49 self .last = {}
50 self .count = 0
51 self .started = time.time()
52 self .server = ThreadingHTTPServer((bind, 0 ), self .handler())
53 self .server.daemon_threads = True
54 self .thread = threading.Thread( target = self .server.serve_forever, daemon = True )
55
56 @ property
57 def port (self):
58 return self .server.server_address[ 1 ]
59
60 def base_url (self, host):
61 return f 'http:// { host } : { self .port }{ self .upstream.path.rstrip( "/" ) } '
62
63 def start (self):
64 self .out_file.parent.mkdir( parents = True , exist_ok = True )
65 self .out_file.write_text( '' )
66 self .thread.start()
67 return self
68
69 def stop (self, timeout = 30 ):
70 """Stop the server. Wait until the open requests are written, so count is final."""
71 self .server.shutdown()
72 with self .idle:
73 self .idle.wait_for( lambda : self .active == 0 , timeout = timeout)
74 self .server.server_close()
75
76 def begin (self):
77 with self .lock:
78 self .active += 1
79
80 def end (self):
81 with self .idle:
82 self .active -= 1
83 self .idle.notify_all()
84
85 def record (self, method, path, body, encoding, status, error_text):
86 row = { 't' : round (time.time() - self .started, 3 ), 'method' : method, 'path' : path, 'status' : status}
87 try :
88 payload = json.loads(decode_body(body, encoding)) if body else None
89 except ( ValueError , OSError , ImportError ):
90 payload, row[ 'bodyBytes' ] = None , len (body)
91 with self .lock:
92 if isinstance (payload, dict ):
93 items = payload.get( 'input' ) if isinstance (payload.get( 'input' ), list ) else []
94 if items:
95 row[ 'inputItems' ] = len (items)
96 # Codex sends its tools as an additional_tools input item, not in the top-level tools field.
97 extra = [t for i in items if isinstance (i, dict ) and i.get( 'type' ) == 'additional_tools'
98 for t in i.get( 'tools' ) or []]
99 if extra:
100 payload = { ** payload, 'tools' : (payload.get( 'tools' ) or []) + extra}
101 # Side tasks, such as a title request, run on their own thread with other tools.
102 metadata = payload.get( 'client_metadata' ) if isinstance (payload.get( 'client_metadata' ), dict ) else {}
103 thread = metadata.get( 'thread_id' ) or payload.get( 'prompt_cache_key' ) or ''
104 for key, value in payload.items():
105 if key == 'input' :
106 continue
107 if key in CHANGING_KEYS :
108 if self .last.get((thread, key)) == value:
109 continue
110 self .last[(thread, key)] = value
111 row[key] = value
112 if error_text:
113 row[ 'error' ] = error_text
114 self .count += 1
115 with self .out_file.open( 'a' ) as f:
116 f.write(json.dumps(row, ensure_ascii = False ) + ' \n ' )
117
118 def handler (self):
119 recorder = self
120
121 class Handler ( BaseHTTPRequestHandler ):
122 protocol_version = 'HTTP/1.1'
123
124 def log_message (self, * args):
125 pass
126
127 def read_body (self):
128 if self .headers.get( 'Content-Length' ):
129 return self .rfile.read( int ( self .headers[ 'Content-Length' ]))
130 if 'chunked' not in ( self .headers.get( 'Transfer-Encoding' ) or '' ).lower():
131 return b ''
132 parts = []
133 while True :
134 size = int ( self .rfile.readline().split( b ';' )[ 0 ].strip() or b '0' , 16 )
135 if not size:
136 self .rfile.readline()
137 return b '' .join(parts)
138 parts.append( self .rfile.read(size))
139 self .rfile.readline()
140
141 def forward (self):
142 recorder.begin()
143 try :
144 self .forward_and_record()
145 finally :
146 recorder.end()
147
148 def forward_and_record (self):
149 body = self .read_body()
150 up = recorder.upstream
151 conn_class = http.client.HTTPSConnection if up.scheme == 'https' else http.client.HTTPConnection
152 conn = conn_class(up.hostname, up.port, timeout = 900 )
153 headers = {k: v for k, v in self .headers.items() if k.lower() not in HOP_HEADERS }
154 headers[ 'Host' ] = up.netloc
155 headers[ 'Content-Length' ] = str ( len (body))
156 status, error_text, replied = 502 , None , False
157 try :
158 conn.request( self .command, self .path, body = body, headers = headers)
159 resp = conn.getresponse()
160 status, replied = resp.status, True
161 self .send_response(resp.status, resp.reason)
162 for key, value in resp.getheaders():
163 if key.lower() not in HOP_HEADERS :
164 self .send_header(key, value)
165 self .send_header( 'Transfer-Encoding' , 'chunked' )
166 self .end_headers()
167 kept = b ''
168 while chunk := resp.read1( 65536 ):
169 if status >= 400 and len (kept) < 4000 :
170 kept += chunk
171 self .wfile.write( b ' %x\r\n%s\r\n ' % ( len (chunk), chunk))
172 self .wfile.flush()
173 self .wfile.write( b '0 \r\n\r\n ' )
174 self .wfile.flush()
175 if kept:
176 error_text = kept[: 4000 ].decode( 'utf-8' , 'replace' )
177 except ( OSError , http.client.HTTPException) as e:
178 error_text = f 'recorder: { type (e). __name__ } '
179 if not replied:
180 try :
181 self .send_error( 502 , 'Recorder could not reach the API' )
182 except OSError :
183 pass
184 self .close_connection = True
185 finally :
186 conn.close()
187 recorder.record( self .command, self .path, body, self .headers.get( 'Content-Encoding' ),
188 status, error_text)
189
190 do_GET = do_POST = do_PUT = do_DELETE = do_PATCH = forward
191
192 return Handler