Claude API实战：如何安全高效地获取和管理Token

1次阅读

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

在开始使用 Claude API 之前，我们需要先理解 OAuth 2.0 的基本工作原理。OAuth 2.0 是一个授权框架，允许第三方应用在用户授权下访问特定资源，而无需直接暴露用户凭据。

Access Token 与 Refresh Token 的区别
Access Token 是短期有效的令牌，用于访问受保护的资源
Refresh Token 是长期有效的令牌，用于获取新的 Access Token
Access Token 通常有效期较短（如 1 小时），Refresh Token 有效期较长（如 30 天）
典型授权码流程
用户被重定向到授权服务器进行认证
授权服务器返回授权码给客户端
客户端使用授权码换取 Access Token 和 Refresh Token
客户端使用 Access Token 访问 API 资源

下面我们通过 Python 代码示例演示完整的授权流程。我们使用 requests 库与 Claude API 交互。

import requests
from base64 import b64encode
from datetime import datetime, timedelta
import json

# 配置你的客户端凭证
CLIENT_ID = 'your_client_id'
CLIENT_SECRET = 'your_client_secret'
REDIRECT_URI = 'https://your-app.com/callback'
AUTH_URL = 'https://api.claude.ai/oauth2/authorize'
TOKEN_URL = 'https://api.claude.ai/oauth2/token'

# 第一步：生成授权 URL
def generate_auth_url():
    params = {
        'response_type': 'code',
        'client_id': CLIENT_ID,
        'redirect_uri': REDIRECT_URI,
        'scope': 'openid profile email',
        'state': 'random_state_string'
    }
    auth_url = requests.Request('GET', AUTH_URL, params=params).prepare().url
    return auth_url

# 第二步：使用授权码换取 Token
def exchange_code_for_token(authorization_code):
    auth_header = b64encode(f"{CLIENT_ID}:{CLIENT_SECRET}".encode()).decode()
    headers = {'Authorization': f'Basic {auth_header}',
        'Content-Type': 'application/x-www-form-urlencoded'
    }
    data = {
        'grant_type': 'authorization_code',
        'code': authorization_code,
        'redirect_uri': REDIRECT_URI
    }

    response = requests.post(TOKEN_URL, headers=headers, data=data)
    response.raise_for_status()

    token_data = response.json()
    token_data['expires_at'] = (datetime.now() + 
                               timedelta(seconds=token_data['expires_in'])).timestamp()

    return token_data

在生产环境中，Token 的安全存储至关重要。以下是几种推荐的存储方案：

AWS Secrets Manager 方案

import boto3
from botocore.exceptions import ClientError

def store_token_in_aws_secrets(token_data, secret_name):
    client = boto3.client('secretsmanager')
    try:
        response = client.create_secret(
            Name=secret_name,
            SecretString=json.dumps(token_data)
        )
    except ClientError as e:
        if e.response['Error']['Code'] == 'ResourceExistsException':
            response = client.update_secret(
                SecretId=secret_name,
                SecretString=json.dumps(token_data)
            )
        else:
            raise
    return response

HashiCorp Vault 方案

import hvac

def store_token_in_vault(token_data, path):
    client = hvac.Client(url='http://localhost:8200', token='your_vault_token')
    client.secrets.kv.v2.create_or_update_secret(
        path=path,
        secret=token_data
    )

为了避免 API 调用因 Token 过期而中断，我们需要实现自动刷新逻辑。

import time

def get_valid_token(token_store):
    token_data = token_store.get_token()

    # 检查 Token 是否即将过期（提前 5 分钟刷新）if datetime.now().timestamp() > (token_data['expires_at'] - 300):
        new_token = refresh_token(token_data['refresh_token'])
        token_store.update_token(new_token)
        return new_token['access_token']

    return token_data['access_token']

def refresh_token(refresh_token):
    auth_header = b64encode(f"{CLIENT_ID}:{CLIENT_SECRET}".encode()).decode()
    headers = {'Authorization': f'Basic {auth_header}',
        'Content-Type': 'application/x-www-form-urlencoded'
    }
    data = {
        'grant_type': 'refresh_token',
        'refresh_token': refresh_token
    }

    response = requests.post(TOKEN_URL, headers=headers, data=data)
    response.raise_for_status()

    token_data = response.json()
    token_data['expires_at'] = (datetime.now() + 
                               timedelta(seconds=token_data['expires_in'])).timestamp()

    return token_data