Setting the file. One moment.
N8n API · N8n:testmax N8n AI · n8n-io/n8n · Skills Docs
Back to the top of the page — line 108
This file
Number 40.11
Position 11 of 24
Type Python
Size 4 KB
Lines 123 scripts/ n8n_api.py
Python · 123 lines · 4 KB
13
REPO_ROOT
=
Path(
__file__
).resolve().parents[
4
]
14 DEFAULT_BASE_URL = 'http://localhost:5678'
15 TOKEN_FILE = Path.home() / '.config' / 'n8n-builder-compare' / 'mcp-token'
16
17
18 def read_env_file (path):
19 """Return KEY=VALUE pairs from a dotenv file. Values are never printed."""
20 values = {}
21 p = Path(path)
22 if not p.exists():
23 return values
24 for line in p.read_text().splitlines():
25 line = line.strip()
26 if line.startswith( 'export ' ):
27 line = line[ 7 :]
28 if not line or line.startswith( '#' ) or '=' not in line:
29 continue
30 key, value = line.split( '=' , 1 )
31 value = value.strip()
32 if value[: 1 ] in ( '"' , "'" ):
33 value = value[ 1 :].split(value[ 0 ], 1 )[ 0 ]
34 else :
35 value = value.split( ' #' , 1 )[ 0 ].strip()
36 values[key.strip()] = value
37 return values
38
39
40 def env_files ():
41 """Config files from lowest to highest priority."""
42 custom = os.environ.get( 'AI_SURFACES_ENV' )
43 return [ REPO_ROOT / '.env.local' , Path(custom) if custom else REPO_ROOT / '.env.ai-surfaces' ]
44
45
46 @cache
47 def config ():
48 """Merged settings: .env.local, then .env.ai-surfaces, then the process environment."""
49 merged = {}
50 for path in env_files():
51 merged.update({k: v for k, v in read_env_file(path).items() if v != '' })
52 merged.update({k: v for k, v in os.environ.items() if v != '' })
53 return merged
54
55
56 def setting ( * names, default = None ):
57 """Return the first name that has a value."""
58 values = config()
59 for name in names:
60 if values.get(name):
61 return values[name]
62 return default
63
64
65 class N8nClient :
66 def __init__ (self):
67 self .base_url = setting( 'N8N_BASE_URL' , default = DEFAULT_BASE_URL ).rstrip( '/' )
68 self .email = setting( 'N8N_EVAL_EMAIL' , default = 'nathan@n8n.io' )
69 self .password = setting( 'N8N_EVAL_PASSWORD' , default = 'PlaywrightTest123' )
70 # The auth cookie is bound to the browser-id header that was sent at login.
71 self .browser_id = str (uuid.uuid4())
72 self .cookies = http.cookiejar.CookieJar()
73 self .opener = urllib.request.build_opener(urllib.request.HTTPCookieProcessor( self .cookies))
74
75 def headers (self, extra = None ):
76 h = { 'content-type' : 'application/json' , 'browser-id' : self .browser_id}
77 h.update(extra or {})
78 return h
79
80 def request (self, method, path, body = None , timeout = 60 ):
81 data = json.dumps(body).encode() if body is not None else None
82 req = urllib.request.Request( self .base_url + path, method = method, data = data, headers = self .headers())
83 try :
84 with self .opener.open(req, timeout = timeout) as res:
85 raw = res.read()
86 except urllib.error.HTTPError as e:
87 detail = e.read()[: 500 ].decode( errors = 'replace' )
88 raise RuntimeError ( f ' { method } { path } -> { e.code } : { detail } ' ) from None
89 parsed = json.loads(raw) if raw else None
90 if isinstance (parsed, dict ) and set (parsed) == { 'data' }:
91 return parsed[ 'data' ]
92 return parsed
93
94 def open_stream (self, path):
95 req = urllib.request.Request(
96 self .base_url + path, headers = self .headers({ 'accept' : 'text/event-stream' })
97 )
98 return self .opener.open(req, timeout = 600 )
99
100 def login (self):
101 self .request( 'POST' , '/rest/login' , { 'emailOrLdapLoginId' : self .email, 'password' : self .password})
102 return self
103
104 def workflow (self, workflow_id):
105 return self .request( 'GET' , f '/rest/workflows/ { workflow_id } ' )
106
107
108 def save_token (token):
109 TOKEN_FILE .parent.mkdir( parents = True , exist_ok = True )
110 TOKEN_FILE .write_text(token)
111 os.chmod( TOKEN_FILE , stat. S_IRUSR | stat. S_IWUSR )
112
113
114 def load_token ():
115 """N8N_MCP_TOKEN from the config wins over the cached token."""
116 token = setting( 'N8N_MCP_TOKEN' )
117 if token:
118 return token
119 return TOKEN_FILE .read_text().strip() if TOKEN_FILE .exists() else None
120
121
122 def is_full_token (value):
123 return isinstance (value, str ) and value.count( '.' ) == 2 and '*' not in value