使用 ChatGPT 登录到 Codex 的完整指南：从认证流程到 API 集成

2次阅读

没有评论

共计 3768 个字符，预计需要花费 10 分钟才能阅读完成。

在集成 OpenAI 的 Codex API 时，开发者常常遇到认证流程复杂、令牌管理困难等问题。尤其是对于中级开发者来说，如何安全高效地实现认证登录是一个不小的挑战。常见的痛点包括：

API 密钥的管理不当导致泄露风险
令牌过期处理逻辑不完善
缺乏有效的错误处理机制
对不同的认证方式选择困惑

OpenAI 提供了多种认证方式，每种方式都有其适用场景：

API Key
最简单直接的方式
适合快速原型开发
安全性较低，需要妥善保管
OAuth 2.0
最安全的认证方式
支持令牌刷新机制
适合生产环境使用
Session Token
适合短期交互场景
不需要长期存储凭证
安全性介于 API Key 和 OAuth 之间

首先需要在 OpenAI 官网获取 API 密钥：

登录 OpenAI 开发者平台
进入 API Keys 管理页面
创建新的 API 密钥
妥善保管密钥，建议使用环境变量存储

使用 OAuth 2.0 时需要实现令牌刷新逻辑：

import requests
from datetime import datetime, timedelta

class TokenManager:
    def __init__(self, client_id, client_secret):
        self.client_id = client_id
        self.client_secret = client_secret
        self.access_token = None
        self.expires_at = None

    def get_token(self):
        if self.access_token and datetime.now() < self.expires_at:
            return self.access_token

        # 令牌过期或不存在，获取新令牌
        token_url = "https://api.openai.com/v1/oauth/token"
        payload = {
            "grant_type": "client_credentials",
            "client_id": self.client_id,
            "client_secret": self.client_secret
        }

        response = requests.post(token_url, data=payload)
        if response.status_code == 200:
            token_data = response.json()
            self.access_token = token_data["access_token"]
            # 设置令牌过期时间（提前 5 分钟刷新）self.expires_at = datetime.now() + timedelta(seconds=token_data["expires_in"] - 300)
            return self.access_token
        else:
            raise Exception("Failed to get access token")

完善的错误处理机制对于稳定运行至关重要：

def make_api_request(url, token_manager):
    try:
        headers = {"Authorization": f"Bearer {token_manager.get_token()}",
            "Content-Type": "application/json"
        }

        response = requests.get(url, headers=headers)

        # 处理 401 未授权错误
        if response.status_code == 401:
            # 尝试刷新令牌后重试
            token_manager.access_token = None
            headers["Authorization"] = f"Bearer {token_manager.get_token()}"
            response = requests.get(url, headers=headers)

        if response.status_code != 200:
            raise Exception(f"API request failed with status {response.status_code}")

        return response.json()

    except Exception as e:
        print(f"Error making API request: {str(e)}")
        raise

import os
from dotenv import load_dotenv

# 加载环境变量
load_dotenv()

class APIClient:
    def __init__(self):
        self.base_url = "https://api.openai.com/v1"
        self.token_manager = TokenManager(os.getenv("OPENAI_CLIENT_ID"),
            os.getenv("OPENAI_CLIENT_SECRET")
        )

    def call_codex(self, prompt):
        url = f"{self.base_url}/codex/completions"
        data = {
            "prompt": prompt,
            "max_tokens": 100
        }

        try:
            headers = {"Authorization": f"Bearer {self.token_manager.get_token()}",
                "Content-Type": "application/json"
            }

            response = requests.post(url, json=data, headers=headers)
            response.raise_for_status()
            return response.json()

        except requests.exceptions.RequestException as e:
            print(f"API call failed: {e}")
            return None

import pickle
import os

class PersistentTokenManager(TokenManager):
    def __init__(self, client_id, client_secret, cache_file="token_cache.pkl"):
        super().__init__(client_id, client_secret)
        self.cache_file = cache_file
        self.load_cache()

    def load_cache(self):
        if os.path.exists(self.cache_file):
            with open(self.cache_file, "rb") as f:
                cache = pickle.load(f)
                self.access_token = cache.get("access_token")
                self.expires_at = cache.get("expires_at")

    def save_cache(self):
        cache = {
            "access_token": self.access_token,
            "expires_at": self.expires_at
        }
        with open(self.cache_file, "wb") as f:
            pickle.dump(cache, f)

    def get_token(self):
        token = super().get_token()
        self.save_cache()
        return token

永远不要将 API 密钥硬编码在代码中
使用环境变量或密钥管理服务（如 AWS Secrets Manager）
令牌缓存应加密存储

始终使用 HTTPS 协议
验证服务器证书
考虑使用双向 TLS 认证

遵循最小权限原则
为不同功能使用不同 API 密钥
定期轮换密钥

实现请求队列和速率限制
考虑使用指数退避算法处理限流错误

import time

def call_api_with_retry(api_call, max_retries=3):
    for attempt in range(max_retries):
        try:
            return api_call()
        except requests.exceptions.HTTPError as e:
            if e.response.status_code == 429:  # 速率限制
                wait_time = 2 ** attempt  # 指数退避
                time.sleep(wait_time)
                continue
            raise
    raise Exception("Max retries exceeded")