FazBrowse GitHub Viewer
|
Trending
|
URL:
|
Home
Tools:
[Download Repo ZIP]
[View Raw Code]
[Original HTTPS Page]
cecli/cecli/onboarding.py at main · cecli-dev/cecli · GitHub
cecli-dev
cecli
Repository navigation
Code
Issues
11
(11)
Pull requests
8
(8)
Actions
Projects
Security and quality
Insights
Expand file tree
Breadcrumbs
cecli
/
cecli
/
onboarding.py
Copy path
More file actions
More file actions
Latest commit
History
History
History
318 lines (285 loc) · 11.3 KB
Breadcrumbs
cecli
/
cecli
/
onboarding.py
Copy path
File metadata and controls
318 lines (285 loc) · 11.3 KB
Raw
Copy raw file
Download raw file
Open symbols panel
Edit and raw actions
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
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
import
http
.
server
import
os
import
socketserver
import
threading
import
time
import
webbrowser
from
urllib
.
parse
import
parse_qs
,
urlparse
import
requests
from
cecli
import
urls
from
cecli
.
decoding
import
safe_open
from
cecli
.
io
import
InputOutput
from
cecli
.
mcp
import
find_available_port
,
generate_pkce_codes
def
check_openrouter_tier
(
api_key
):
"""
Checks if the user is on a free tier for OpenRouter.
Args:
api_key: The OpenRouter API key to check.
Returns:
A boolean indicating if the user is on a free tier (True) or paid tier (False).
Returns True if the check fails.
"""
try
:
response
=
requests
.
get
(
"https://openrouter.ai/api/v1/auth/key"
,
headers
=
{
"Authorization"
:
f"Bearer
{
api_key
}
"
},
timeout
=
5
,
)
response
.
raise_for_status
()
data
=
response
.
json
()
return
data
.
get
(
"data"
, {}).
get
(
"is_free_tier"
,
True
)
except
Exception
:
return
True
def
try_to_select_default_model
():
"""
Attempts to select a default model based on available API keys.
Checks OpenRouter tier status to select appropriate model.
Returns:
The name of the selected model, or None if no suitable default is found.
"""
openrouter_key
=
os
.
environ
.
get
(
"OPENROUTER_API_KEY"
)
if
openrouter_key
:
is_free_tier
=
check_openrouter_tier
(
openrouter_key
)
if
is_free_tier
:
return
"openrouter/nvidia/nemotron-3-ultra-550b-a55b:free"
else
:
return
"openrouter/anthropic/claude-sonnet-5"
model_key_pairs
=
[
(
"ANTHROPIC_API_KEY"
,
"anthropic/claude-sonnet-5"
),
(
"DEEPSEEK_API_KEY"
,
"deepseek-v4-flash"
),
(
"OPENAI_API_KEY"
,
"gpt-5.6-luna"
),
(
"GEMINI_API_KEY"
,
"gemini/gemini-3.8-flash"
),
(
"VERTEXAI_PROJECT"
,
"gemini/gemini-3.8-flash"
),
]
for
env_key
,
model_name
in
model_key_pairs
:
api_key_value
=
os
.
environ
.
get
(
env_key
)
if
api_key_value
:
return
model_name
return
None
async
def
offer_openrouter_oauth
(
io
):
"""
Offers OpenRouter OAuth flow to the user if no API keys are found.
Args:
io: The InputOutput object for user interaction.
Returns:
True if authentication was successful, False otherwise.
"""
io
.
tool_output
(
"OpenRouter provides free and paid access to many LLMs."
)
if
await
io
.
confirm_ask
(
"Login to OpenRouter or create a free account?"
,
default
=
"y"
,
acknowledge
=
True
):
openrouter_key
=
start_openrouter_oauth_flow
(
io
)
if
openrouter_key
:
os
.
environ
[
"OPENROUTER_API_KEY"
]
=
openrouter_key
return
True
io
.
tool_error
(
"OpenRouter authentication did not complete successfully."
)
return
False
async
def
select_default_model
(
args
,
io
):
"""
Selects a default model based on available API keys if no model is specified.
Launches the inline onboarding wizard when no model or API keys are found.
When ``args.configure_provider`` is set, the onboarding wizard is always launched
first, even if a model or API keys are already configured.
Args:
args: The command line arguments object.
io: The InputOutput object for user interaction.
Returns:
The name of the selected model, or None if no suitable default is found.
"""
from
cecli
.
helpers
.
onboarding
import
run_onboarding
configure_provider
=
getattr
(
args
,
"configure_provider"
,
False
)
onboarded
=
False
if
configure_provider
:
onboarded
=
True
model
=
await
run_onboarding
(
io
)
if
model
:
return
model
if
args
.
model
:
return
args
.
model
model
=
try_to_select_default_model
()
if
model
:
io
.
tool_warning
(
f"Using
{
model
}
model with API key from environment."
)
return
model
if
onboarded
:
await
io
.
offer_url
(
urls
.
models_and_keys
,
"Open documentation URL for more info?"
)
return
None
no_model_msg
=
"No LLM model was specified and no API keys were provided."
io
.
tool_warning
(
no_model_msg
)
model
=
await
run_onboarding
(
io
)
if
model
:
return
model
await
io
.
offer_url
(
urls
.
models_and_keys
,
"Open documentation URL for more info?"
)
# Function to exchange the authorization code for an API key
def
exchange_code_for_key
(
code
,
code_verifier
,
io
):
try
:
response
=
requests
.
post
(
"https://openrouter.ai/api/v1/auth/keys"
,
headers
=
{
"Content-Type"
:
"application/json"
},
json
=
{
"code"
:
code
,
"code_verifier"
:
code_verifier
,
"code_challenge_method"
:
"S256"
},
timeout
=
30
,
)
response
.
raise_for_status
()
data
=
response
.
json
()
api_key
=
data
.
get
(
"key"
)
if
not
api_key
:
io
.
tool_error
(
"Error: 'key' not found in OpenRouter response."
)
io
.
tool_error
(
f"Response:
{
response
.
text
}
"
)
return
None
return
api_key
except
requests
.
exceptions
.
Timeout
:
io
.
tool_error
(
"Error: Request to OpenRouter timed out during code exchange."
)
return
None
except
requests
.
exceptions
.
HTTPError
as
e
:
io
.
tool_error
(
"Error exchanging code for OpenRouter key:"
f"
{
e
.
response
.
status_code
}
{
e
.
response
.
reason
}
"
)
io
.
tool_error
(
f"Response:
{
e
.
response
.
text
}
"
)
return
None
except
requests
.
exceptions
.
RequestException
as
e
:
io
.
tool_error
(
f"Error exchanging code for OpenRouter key:
{
e
}
"
)
return
None
except
Exception
as
e
:
io
.
tool_error
(
f"Unexpected error during code exchange:
{
e
}
"
)
return
None
def
start_openrouter_oauth_flow
(
io
):
"""Initiates the OpenRouter OAuth PKCE flow using a local server."""
port
=
find_available_port
()
if
not
port
:
io
.
tool_error
(
"Could not find an available port between 8484 and 8584."
)
io
.
tool_error
(
"Please ensure a port in this range is free, or configure manually."
)
return
None
callback_url
=
f"http://localhost:
{
port
}
/callback/cecli"
auth_code
=
None
server_error
=
None
server_started
=
threading
.
Event
()
shutdown_server
=
threading
.
Event
()
class
OAuthCallbackHandler
(
http
.
server
.
SimpleHTTPRequestHandler
):
def
do_GET
(
self
):
nonlocal
auth_code
,
server_error
# noqa
parsed_path
=
urlparse
(
self
.
path
)
if
parsed_path
.
path
==
"/callback/cecli"
:
query_params
=
parse_qs
(
parsed_path
.
query
)
if
"code"
in
query_params
:
auth_code
=
query_params
[
"code"
][
0
]
self
.
send_response
(
200
)
self
.
send_header
(
"Content-type"
,
"text/html"
)
self
.
end_headers
()
self
.
wfile
.
write
(
b"<html><body><h1>Success!</h1><p>cecli has received the authentication"
b" code. You can close this browser tab.</p></body></html>"
)
shutdown_server
.
set
()
else
:
self
.
send_response
(
302
)
self
.
send_header
(
"Location"
,
urls
.
website
)
self
.
end_headers
()
else
:
self
.
send_response
(
302
)
self
.
send_header
(
"Location"
,
urls
.
website
)
self
.
end_headers
()
self
.
wfile
.
write
(
b"Not Found"
)
def
log_message
(
self
,
format
,
*
args
):
pass
def
run_server
():
nonlocal
server_error
# noqa
try
:
with
socketserver
.
TCPServer
((
"localhost"
,
port
),
OAuthCallbackHandler
)
as
httpd
:
io
.
tool_output
(
f"Temporary server listening on
{
callback_url
}
"
,
log_only
=
True
)
server_started
.
set
()
while
not
shutdown_server
.
is_set
():
httpd
.
handle_request
()
time
.
sleep
(
0.1
)
io
.
tool_output
(
"Shutting down temporary server."
,
log_only
=
True
)
except
Exception
as
e
:
server_error
=
f"Failed to start or run temporary server:
{
e
}
"
# noqa
server_started
.
set
()
shutdown_server
.
set
()
server_thread
=
threading
.
Thread
(
target
=
run_server
,
daemon
=
True
)
server_thread
.
start
()
if
not
server_started
.
wait
(
timeout
=
5
):
io
.
tool_error
(
"Temporary authentication server failed to start in time."
)
shutdown_server
.
set
()
server_thread
.
join
(
timeout
=
1
)
return
None
if
server_error
:
io
.
tool_error
(
server_error
)
shutdown_server
.
set
()
server_thread
.
join
(
timeout
=
1
)
return
None
code_verifier
,
code_challenge
=
generate_pkce_codes
()
auth_url_base
=
"https://openrouter.ai/auth"
auth_params
=
{
"callback_url"
:
callback_url
,
"code_challenge"
:
code_challenge
,
"code_challenge_method"
:
"S256"
,
}
auth_url
=
f"
{
auth_url_base
}
?
{
'&'
.
join
(
f'
{
k
}
=
{
v
}
'
for
k
,
v
in
auth_params
.
items
())
}
"
io
.
tool_output
(
"
\n
Please open this URL in your browser to connect cecli with OpenRouter:"
)
io
.
tool_output
()
print
(
auth_url
)
MINUTES
=
5
io
.
tool_output
(
f"
\n
Waiting up to
{
MINUTES
}
minutes for you to finish in the browser..."
)
io
.
tool_output
(
"Use Control-C to interrupt."
)
try
:
webbrowser
.
open
(
auth_url
)
except
Exception
:
pass
interrupted
=
False
try
:
shutdown_server
.
wait
(
timeout
=
MINUTES
*
60
)
except
KeyboardInterrupt
:
io
.
tool_warning
(
"
\n
OAuth flow interrupted."
)
interrupted
=
True
shutdown_server
.
set
()
server_thread
.
join
(
timeout
=
1
)
if
interrupted
:
return
None
if
server_error
:
io
.
tool_error
(
f"Authentication failed:
{
server_error
}
"
)
return
None
if
not
auth_code
:
io
.
tool_error
(
"Authentication with OpenRouter failed."
)
return
None
io
.
tool_output
(
"Completing authentication..."
)
api_key
=
exchange_code_for_key
(
auth_code
,
code_verifier
,
io
)
if
api_key
:
os
.
environ
[
"OPENROUTER_API_KEY"
]
=
api_key
try
:
config_dir
=
os
.
path
.
expanduser
(
"~/.cecli"
)
os
.
makedirs
(
config_dir
,
exist_ok
=
True
)
key_file
=
os
.
path
.
join
(
config_dir
,
"oauth-keys.env"
)
with
safe_open
(
key_file
,
"a"
)
as
f
:
f
.
write
(
f'OPENROUTER_API_KEY="
{
api_key
}
"
\n
'
)
io
.
tool_warning
(
"cecli will load the OpenRouter key automatically in future sessions."
)
io
.
tool_output
()
return
api_key
except
Exception
as
e
:
io
.
tool_error
(
f"Successfully obtained key, but failed to save it to file:
{
e
}
"
)
io
.
tool_warning
(
"Set OPENROUTER_API_KEY environment variable for this session only."
)
return
api_key
else
:
io
.
tool_error
(
"Authentication with OpenRouter failed."
)
return
None
def
main
():
"""Main function to test the OpenRouter OAuth flow."""
print
(
"Starting OpenRouter OAuth flow test..."
)
io
=
InputOutput
(
pretty
=
True
,
yes
=
False
,
input_history_file
=
None
,
chat_history_file
=
None
,
tool_output_color
=
"BLUE"
,
tool_error_color
=
"RED"
,
)
if
"OPENROUTER_API_KEY"
in
os
.
environ
:
print
(
"Warning: OPENROUTER_API_KEY is already set in environment."
)
api_key
=
start_openrouter_oauth_flow
(
io
)
if
api_key
:
print
(
"
\n
OAuth flow completed successfully!"
)
print
(
f"Obtained API Key (first 5 chars):
{
api_key
[:
5
]
}
..."
)
else
:
print
(
"
\n
OAuth flow failed or was cancelled."
)
print
(
"
\n
OpenRouter OAuth flow test finished."
)
if
__name__
==
"__main__"
:
main
()
Back
|
FazBrowse Home
|
New Git URL