""" title: GitHub Copilot Pipe author: dragon-fish author_url: https://github.com/dragon-fish funding_url: https://afdian.com/a/dragon-fish version: 0.1.0 license: MIT """ import os import time from pydantic import BaseModel, Field import requests from dataclasses import dataclass from typing import Iterator, List, Optional, Dict, Any, Union import httpx from openai import OpenAI """ GithubCopilotAI like OpenAI client author: dragon-fish license: MIT see: https://github.com/dragon-fish/github-copilot-ai """ @dataclass class CopilotInternalUser: access_type_sku: str analytics_tracking_id: str assigned_date: str can_signup_for_limited: bool chat_enabled: bool copilot_plan: str organization_login_list: Any organization_list: Any @dataclass class CopilotInternalAuth: annotations_enabled: bool chat_enabled: bool chat_jetbrains_enabled: bool code_quote_enabled: bool code_review_enabled: bool codesearch: bool copilotignore_enabled: bool endpoints: Dict[str, str] expires_at: int individual: bool limited_user_quotas: Optional[Any] limited_user_reset_date: Optional[Union[str, Any]] prompt_8k: bool public_suggestions: str refresh_in: int sku: str snippy_load_test_enabled: bool telemetry: str token: str tracking_id: str vsc_electron_fetcher_v2: bool xcode: bool xcode_chat: bool class GithubCopilotAI(OpenAI): # Copilot internal state _copilot_internal_user: Optional[CopilotInternalUser] = None _copilot_internal_auth: Optional[CopilotInternalAuth] = None # Default headers for Copilot integration default_headers: Dict[str, str] = { "copilot-integration-id": "vscode-chat", "editor-plugin-version": "copilot-chat/0.28.0", "editor-version": "vscode/1.100.0-insider", "openai-intent": "conversation-panel", "user-agent": "GitHubCopilotChat/0.28.0", } def __init__( self, api_key: Optional[str] = None, copilot_plan: str = "default", **kwargs: Any, ): # Determine base URL by plan base_url = self.get_base_url_by_plan(copilot_plan) # Create HTTPX client with default headers and request hook client = httpx.Client( base_url=base_url, headers=self.default_headers, event_hooks={"request": [self._on_request]}, ) # Initialize parent OpenAI client with custom HTTP client super().__init__(api_key=api_key, http_client=client, **kwargs) # Store base_url for dynamic plan updates self.base_url = base_url @staticmethod def get_base_url_by_plan(copilot_plan: str = "default") -> str: if copilot_plan == "enterprise": return "https://api.enterprise.githubcopilot.com/" elif copilot_plan == "individual": return "https://api.individual.githubcopilot.com/" else: return "https://api.githubcopilot.com/" def _on_request(self, request: httpx.Request) -> None: # Attach dynamic authorization header for Copilot internal API auth = self.get_copilot_internal_auth() if auth is None: raise ValueError("No Copilot internal auth available") request.headers["Authorization"] = f"Bearer {auth.token}" def get_copilot_internal_user(self) -> Optional[CopilotInternalUser]: # Return cached user if still valid if self._copilot_internal_user and self._copilot_internal_user.assigned_date: return self._copilot_internal_user # Fetch from GitHub API headers = { "authorization": f"Bearer {self.api_key}", "accept": "application/json", } headers.update(self.default_headers) resp = requests.get( "https://api.github.com/copilot_internal/user", headers=headers ) data = resp.json() user = CopilotInternalUser(**data) return self.set_copilot_internal_user(user) def set_copilot_internal_user( self, user: Optional[CopilotInternalUser] ) -> Optional[CopilotInternalUser]: if not user: self._copilot_internal_user = None return None if not user.assigned_date: raise ValueError("Invalid payload for CopilotInternalUser", user) self._copilot_internal_user = user # Update base_url based on user plan self.base_url = self.get_base_url_by_plan( user.copilot_plan or "default") return self._copilot_internal_user def get_copilot_internal_auth(self) -> Optional[CopilotInternalAuth]: # Return cached auth if not expired if ( self._copilot_internal_auth and self._copilot_internal_auth.expires_at > time.time() ): return self._copilot_internal_auth # Fetch new token headers = { "authorization": f"Bearer {self.api_key}", "accept": "application/json", } headers.update(self.default_headers) resp = requests.get( "https://api.github.com/copilot_internal/v2/token", headers=headers ) data = resp.json() auth = CopilotInternalAuth(**data) return self.set_copilot_internal_auth(auth) def set_copilot_internal_auth( self, auth: Optional[CopilotInternalAuth] ) -> Optional[CopilotInternalAuth]: if not auth: self._copilot_internal_auth = None return None if not auth.token or not auth.expires_at: raise ValueError("Invalid payload for CopilotInternalAuth", auth) if time.time() > auth.expires_at: self._copilot_internal_auth = None return None self._copilot_internal_auth = auth return self._copilot_internal_auth """ Main Pipe for open-webui """ class Pipe: _client: GithubCopilotAI class Valves(BaseModel): COPILOT_API_KEY: str = Field( default="", description="GitHub Copilot API Key, begin with ghu_" ) def __init__(self): self.id = "copilot" self.type = "manifold" self.name = "Copilot: " self.valves = self.Valves( **{ "COPILOT_API_KEY": os.getenv("COPILOT_API_KEY", ""), } ) self._client = GithubCopilotAI( api_key=self.valves.COPILOT_API_KEY, ) def get_all_models(self): if not self.valves.COPILOT_API_KEY: return [ { "id": "error", "name": "COPILOT_API_KEY is not set. Please update the API Key in the valves.", } ] self._client.api_key = self.valves.COPILOT_API_KEY try: models = self._client.models.list() return [ { "id": model.id, "name": model.id, } for model in models ] except Exception as e: return [ { "id": "error", "name": f"Could not fetch models from Copilot: {str(e)}", } ] def pipes(self) -> List[dict]: return self.get_all_models() def pipe(self, body: dict) -> Union[str, Iterator[str]]: if not self.valves.COPILOT_API_KEY: return "Error: COPILOT_API_KEY is not set" try: self._client.api_key = self.valves.COPILOT_API_KEY # 寻找model_id中的第一个 `.`，去掉前缀，比如 "copilot.foo" -> "foo" if "." in body["model"]: body["model"] = body["model"].split(".")[1] stream = body.get("stream", False) if stream: def stream_generator(): response = self._client.chat.completions.create(**body) for chunk in response: if chunk.choices[0].delta.content: yield chunk.choices[0].delta.content return stream_generator() else: response = self._client.chat.completions.create(**body) if response.choices and response.choices[0].message.content: return response.choices[0].message.content else: return "No content returned from Copilot API" except Exception as e: return f"Error: {e}"