| 1 | """GitHub Copilot OAuth-backed provider.""" |
| 2 | |
| 3 | from __future__ import annotations |
| 4 | |
| 5 | import time |
| 6 | import webbrowser |
| 7 | from collections.abc import Callable |
| 8 | |
| 9 | import httpx |
| 10 | from oauth_cli_kit.models import OAuthToken |
| 11 | from oauth_cli_kit.storage import FileTokenStorage |
| 12 | |
| 13 | from nanobot.providers.openai_compat_provider import OpenAICompatProvider |
| 14 | |
| 15 | DEFAULT_GITHUB_DEVICE_CODE_URL = "https://github.com/login/device/code" |
| 16 | DEFAULT_GITHUB_ACCESS_TOKEN_URL = "https://github.com/login/oauth/access_token" |
| 17 | DEFAULT_GITHUB_USER_URL = "https://api.github.com/user" |
| 18 | DEFAULT_COPILOT_TOKEN_URL = "https://api.github.com/copilot_internal/v2/token" |
| 19 | DEFAULT_COPILOT_BASE_URL = "https://api.githubcopilot.com" |
| 20 | GITHUB_COPILOT_CLIENT_ID = "Iv1.b507a08c87ecfe98" |
| 21 | GITHUB_COPILOT_SCOPE = "read:user" |
| 22 | TOKEN_FILENAME = "github-copilot.json" |
| 23 | TOKEN_APP_NAME = "nanobot" |
| 24 | USER_AGENT = "nanobot/0.1" |
| 25 | EDITOR_VERSION = "vscode/1.99.0" |
| 26 | EDITOR_PLUGIN_VERSION = "copilot-chat/0.26.0" |
| 27 | _EXPIRY_SKEW_SECONDS = 60 |
| 28 | _LONG_LIVED_TOKEN_SECONDS = 315360000 |
| 29 | |
| 30 | |
| 31 | def _storage() -> FileTokenStorage: |
| 32 | return FileTokenStorage( |
| 33 | token_filename=TOKEN_FILENAME, |
| 34 | app_name=TOKEN_APP_NAME, |
| 35 | import_codex_cli=False, |
| 36 | ) |
| 37 | |
| 38 | |
| 39 | def _copilot_headers(token: str) -> dict[str, str]: |
| 40 | return { |
| 41 | "Authorization": f"token {token}", |
| 42 | "Accept": "application/json", |
| 43 | "User-Agent": USER_AGENT, |
| 44 | "Editor-Version": EDITOR_VERSION, |
| 45 | "Editor-Plugin-Version": EDITOR_PLUGIN_VERSION, |
| 46 | } |
| 47 | |
| 48 | |
| 49 | def _load_github_token() -> OAuthToken | None: |
| 50 | token = _storage().load() |
| 51 | if not token or not token.access: |
| 52 | return None |
| 53 | return token |
| 54 | |
| 55 | |
| 56 | def get_github_copilot_login_status() -> OAuthToken | None: |
| 57 | """Return the persisted GitHub OAuth token if available.""" |
| 58 | return _load_github_token() |
| 59 | |
| 60 | |
| 61 | def login_github_copilot( |
| 62 | print_fn: Callable[[str], None] | None = None, |
| 63 | prompt_fn: Callable[[str], str] | None = None, |
| 64 | ) -> OAuthToken: |
| 65 | """Run GitHub device flow and persist the GitHub OAuth token used for Copilot.""" |
| 66 | del prompt_fn |
| 67 | printer = print_fn or print |
| 68 | timeout = httpx.Timeout(20.0, connect=20.0) |
| 69 | |
| 70 | with httpx.Client(timeout=timeout, follow_redirects=True, trust_env=True) as client: |
| 71 | response = client.post( |
| 72 | DEFAULT_GITHUB_DEVICE_CODE_URL, |
| 73 | headers={"Accept": "application/json", "User-Agent": USER_AGENT}, |
| 74 | data={"client_id": GITHUB_COPILOT_CLIENT_ID, "scope": GITHUB_COPILOT_SCOPE}, |
| 75 | ) |
| 76 | response.raise_for_status() |
| 77 | payload = response.json() |
| 78 | |
| 79 | device_code = str(payload["device_code"]) |
| 80 | user_code = str(payload["user_code"]) |
| 81 | verify_url = str(payload.get("verification_uri") or payload.get("verification_uri_complete") or "") |
| 82 | verify_complete = str(payload.get("verification_uri_complete") or verify_url) |
| 83 | interval = max(1, int(payload.get("interval") or 5)) |
| 84 | expires_in = int(payload.get("expires_in") or 900) |
| 85 | |
| 86 | printer(f"Open: {verify_url}") |
| 87 | printer(f"Code: {user_code}") |
| 88 | if verify_complete: |
| 89 | try: |
| 90 | webbrowser.open(verify_complete) |
| 91 | except Exception: |
| 92 | pass |
| 93 | |
| 94 | deadline = time.time() + expires_in |
| 95 | current_interval = interval |
| 96 | access_token = None |
| 97 | token_expires_in = _LONG_LIVED_TOKEN_SECONDS |
| 98 | while time.time() < deadline: |
| 99 | poll = client.post( |
| 100 | DEFAULT_GITHUB_ACCESS_TOKEN_URL, |
| 101 | headers={"Accept": "application/json", "User-Agent": USER_AGENT}, |
| 102 | data={ |
| 103 | "client_id": GITHUB_COPILOT_CLIENT_ID, |
| 104 | "device_code": device_code, |
| 105 | "grant_type": "urn:ietf:params:oauth:grant-type:device_code", |
| 106 | }, |
| 107 | ) |
| 108 | poll.raise_for_status() |
| 109 | poll_payload = poll.json() |
| 110 | |
| 111 | access_token = poll_payload.get("access_token") |
| 112 | if access_token: |
| 113 | token_expires_in = int(poll_payload.get("expires_in") or _LONG_LIVED_TOKEN_SECONDS) |
| 114 | break |
| 115 | |
| 116 | error = poll_payload.get("error") |
| 117 | if error == "authorization_pending": |
| 118 | time.sleep(current_interval) |
| 119 | continue |
| 120 | if error == "slow_down": |
| 121 | current_interval += 5 |
| 122 | time.sleep(current_interval) |
| 123 | continue |
| 124 | if error == "expired_token": |
| 125 | raise RuntimeError("GitHub device code expired. Please run login again.") |
| 126 | if error == "access_denied": |
| 127 | raise RuntimeError("GitHub device flow was denied.") |
| 128 | if error: |
| 129 | desc = poll_payload.get("error_description") or error |
| 130 | raise RuntimeError(str(desc)) |
| 131 | time.sleep(current_interval) |
| 132 | else: |
| 133 | raise RuntimeError("GitHub device flow timed out.") |
| 134 | |
| 135 | user = client.get( |
| 136 | DEFAULT_GITHUB_USER_URL, |
| 137 | headers={ |
| 138 | "Authorization": f"Bearer {access_token}", |
| 139 | "Accept": "application/vnd.github+json", |
| 140 | "User-Agent": USER_AGENT, |
| 141 | }, |
| 142 | ) |
| 143 | user.raise_for_status() |
| 144 | user_payload = user.json() |
| 145 | account_id = user_payload.get("login") or str(user_payload.get("id") or "") or None |
| 146 | |
| 147 | expires_ms = int((time.time() + token_expires_in) * 1000) |
| 148 | token = OAuthToken( |
| 149 | access=str(access_token), |
| 150 | refresh="", |
| 151 | expires=expires_ms, |
| 152 | account_id=str(account_id) if account_id else None, |
| 153 | ) |
| 154 | _storage().save(token) |
| 155 | return token |
| 156 | |
| 157 | |
| 158 | class GitHubCopilotProvider(OpenAICompatProvider): |
| 159 | """Provider that exchanges a stored GitHub OAuth token for Copilot access tokens.""" |
| 160 | |
| 161 | def __init__(self, default_model: str = "github-copilot/gpt-4.1"): |
| 162 | from nanobot.providers.registry import find_by_name |
| 163 | |
| 164 | self._copilot_access_token: str | None = None |
| 165 | self._copilot_expires_at: float = 0.0 |
| 166 | super().__init__( |
| 167 | api_key="no-key", |
| 168 | api_base=DEFAULT_COPILOT_BASE_URL, |
| 169 | default_model=default_model, |
| 170 | extra_headers={ |
| 171 | "Editor-Version": EDITOR_VERSION, |
| 172 | "Editor-Plugin-Version": EDITOR_PLUGIN_VERSION, |
| 173 | "User-Agent": USER_AGENT, |
| 174 | }, |
| 175 | spec=find_by_name("github_copilot"), |
| 176 | ) |
| 177 | |
| 178 | async def _get_copilot_access_token(self) -> str: |
| 179 | now = time.time() |
| 180 | if self._copilot_access_token and now < self._copilot_expires_at - _EXPIRY_SKEW_SECONDS: |
| 181 | return self._copilot_access_token |
| 182 | |
| 183 | github_token = _load_github_token() |
| 184 | if not github_token or not github_token.access: |
| 185 | raise RuntimeError("GitHub Copilot is not logged in. Run: nanobot provider login github-copilot") |
| 186 | |
| 187 | timeout = httpx.Timeout(20.0, connect=20.0) |
| 188 | async with httpx.AsyncClient(timeout=timeout, follow_redirects=True, trust_env=True) as client: |
| 189 | response = await client.get( |
| 190 | DEFAULT_COPILOT_TOKEN_URL, |
| 191 | headers=_copilot_headers(github_token.access), |
| 192 | ) |
| 193 | response.raise_for_status() |
| 194 | payload = response.json() |
| 195 | |
| 196 | token = payload.get("token") |
| 197 | if not token: |
| 198 | raise RuntimeError("GitHub Copilot token exchange returned no token.") |
| 199 | |
| 200 | expires_at = payload.get("expires_at") |
| 201 | if isinstance(expires_at, (int, float)): |
| 202 | self._copilot_expires_at = float(expires_at) |
| 203 | else: |
| 204 | refresh_in = payload.get("refresh_in") or 1500 |
| 205 | self._copilot_expires_at = time.time() + int(refresh_in) |
| 206 | self._copilot_access_token = str(token) |
| 207 | return self._copilot_access_token |
| 208 | |
| 209 | async def _refresh_client_api_key(self) -> str: |
| 210 | token = await self._get_copilot_access_token() |
| 211 | self.api_key = token |
| 212 | self._client.api_key = token |
| 213 | return token |
| 214 | |
| 215 | async def chat( |
| 216 | self, |
| 217 | messages: list[dict[str, object]], |
| 218 | tools: list[dict[str, object]] | None = None, |
| 219 | model: str | None = None, |
| 220 | max_tokens: int = 4096, |
| 221 | temperature: float = 0.7, |
| 222 | reasoning_effort: str | None = None, |
| 223 | tool_choice: str | dict[str, object] | None = None, |
| 224 | ): |
| 225 | await self._refresh_client_api_key() |
| 226 | return await super().chat( |
| 227 | messages=messages, |
| 228 | tools=tools, |
| 229 | model=model, |
| 230 | max_tokens=max_tokens, |
| 231 | temperature=temperature, |
| 232 | reasoning_effort=reasoning_effort, |
| 233 | tool_choice=tool_choice, |
| 234 | ) |
| 235 | |
| 236 | async def chat_stream( |
| 237 | self, |
| 238 | messages: list[dict[str, object]], |
| 239 | tools: list[dict[str, object]] | None = None, |
| 240 | model: str | None = None, |
| 241 | max_tokens: int = 4096, |
| 242 | temperature: float = 0.7, |
| 243 | reasoning_effort: str | None = None, |
| 244 | tool_choice: str | dict[str, object] | None = None, |
| 245 | on_content_delta: Callable[[str], None] | None = None, |
| 246 | ): |
| 247 | await self._refresh_client_api_key() |
| 248 | return await super().chat_stream( |
| 249 | messages=messages, |
| 250 | tools=tools, |
| 251 | model=model, |
| 252 | max_tokens=max_tokens, |
| 253 | temperature=temperature, |
| 254 | reasoning_effort=reasoning_effort, |
| 255 | tool_choice=tool_choice, |
| 256 | on_content_delta=on_content_delta, |
| 257 | ) |
| 258 |