Amazon Q to API Bridge - Main Service

Star History

✨ 核心特性

API 兼容性

账号管理

负载与监控

网络与安全

🚀 部署

方式一:Docker Compose

# 1. 复制环境变量配置
cp .env.example .env
# 2. 编辑 .env 文件(可选)
# 配置 OPENAI_KEYS、MAX_ERROR_COUNT 等
# 3. 启动服务
docker-compose up -d
# 4. 查看日志
docker-compose logs -f
# 5. 停止服务
docker-compose down

方式二:本地部署

1. 安装依赖

# 安装 uv
pip install uv
# 创建虚拟环境并安装依赖
uv venv
uv pip install -r requirements.txt

2. 配置环境变量

# 复制示例配置
cp .env.example .env
# 根据需要编辑 .env 文件

# 数据库连接URL(留空使用本地SQLite)
# PostgreSQL: DATABASE_URL="postgres://user:password@host:5432/dbname?sslmode=require"
# MySQL: DATABASE_URL="mysql://user:password@host:3306/dbname"
DATABASE_URL=""
# OpenAI 风格 API Key 白名单(仅用于授权,与账号无关)
# 多个用逗号分隔,例如:OPENAI_KEYS="key1,key2,key3"
# 留空则为开发模式,不校验 Authorization
OPENAI_KEYS=""
# Token 计数倍率(影响 /v1/messages/count_tokens 和 /v1/messages 的输入 token 统计)
# 默认值为 1.0,可根据实际需要调整(如设置为 1.5 表示返回 1.5 倍的 token 数)
TOKEN_COUNT_MULTIPLIER="1.0"
# 出错次数阈值,超过此值自动禁用账号
MAX_ERROR_COUNT=100
# HTTP代理设置(留空不使用代理)
# 例如:HTTP_PROXY="http://127.0.0.1:7890"
HTTP_PROXY=""
# 管理控制台开关(默认启用)
# 设置为 "false" 或 "0" 可禁用管理控制台和相关API端点
ENABLE_CONSOLE="true"
# 管理控制台登录密码(默认 "admin")
# 用于访问管理控制台的密码,会话有效期为30天
ADMIN_PASSWORD="admin"
# 主服务端口(默认 8000)
PORT=8000

3. 启动服务

# 启动服务 (带热重载)
uvicorn app:app --host 0.0.0.0 --port 8000 --reload

📖 使用指南

管理控制台登录

账号管理

方式一:Web 控制台(推荐)

方式二:URL 登录(最简单)

curl -X POST http://localhost:8000/v2/auth/start \
 -H "Content-Type: application/json" \
 -d '{"label": "我的账号", "enabled": true}'

curl -X POST http://localhost:8000/v2/auth/claim/{authId}

方式三:REST API 手动管理

# 登录并获取 token
TOKEN=$(curl -X POST http://localhost:8000/api/login \
 -H "Content-Type: application/json" \
 -d '{"password": "admin"}' \
 | jq -r '.token')

curl -X POST http://localhost:8000/v2/accounts \
 -H "Content-Type: application/json" \
 -H "Authorization: Bearer $TOKEN" \
 -d '{
 "label": "手动创建的账号",
 "clientId": "your-client-id",
 "clientSecret": "your-client-secret",
 "refreshToken": "your-refresh-token",
 "enabled": true
 }'

curl http://localhost:8000/v2/accounts \
 -H "Authorization: Bearer $TOKEN"

curl -X PATCH http://localhost:8000/v2/accounts/{account_id} \
 -H "Content-Type: application/json" \
 -H "Authorization: Bearer $TOKEN" \
 -d '{"enabled": false}'

curl -X POST http://localhost:8000/v2/accounts/{account_id}/refresh \
 -H "Authorization: Bearer $TOKEN"

curl -X DELETE http://localhost:8000/v2/accounts/{account_id} \
 -H "Authorization: Bearer $TOKEN"

OpenAI 兼容 API

非流式请求

curl -X POST http://localhost:8000/v1/chat/completions \
 -H "Content-Type: application/json" \
 -H "Authorization: Bearer your-api-key" \
 -d '{
 "model": "claude-sonnet-4",
 "stream": false,
 "messages": [
 {"role": "system", "content": "你是一个乐于助人的助手"},
 {"role": "user", "content": "你好,请讲一个简短的故事"}
 ]
 }'

流式请求(SSE)

curl -N -X POST http://localhost:8000/v1/chat/completions \
 -H "Content-Type: application/json" \
 -H "Authorization: Bearer your-api-key" \
 -d '{
 "model": "claude-sonnet-4",
 "stream": true,
 "messages": [
 {"role": "user", "content": "讲一个笑话"}
 ]
 }'

Python 示例

import openai
client = openai.OpenAI(
 base_url="http://localhost:8000/v1",
 api_key="your-api-key" # 如果配置了 OPENAI_KEYS
)
response = client.chat.completions.create(
 model="claude-sonnet-4",
 messages=[
 {"role": "user", "content": "你好"}
 ]
)
print(response.choices[0].message.content)

Claude Messages API

基础文本对话

curl -X POST http://localhost:8000/v1/messages \
 -H "Content-Type: application/json" \
 -H "x-api-key: your-api-key" \
 -d '{
 "model": "claude-sonnet-4.5",
 "max_tokens": 1024,
 "messages": [
 {"role": "user", "content": "你好"}
 ]
 }'

Python SDK 示例

from anthropic import Anthropic
client = Anthropic(
 base_url="http://localhost:8000/v1",
 api_key="your-api-key"
)
# 基础对话
message = client.messages.create(
 model="claude-sonnet-4.5",
 max_tokens=1024,
 messages=[
 {"role": "user", "content": "你好"}
 ]
)
print(message.content[0].text)
# 流式响应
with client.messages.stream(
 model="claude-sonnet-4.5",
 max_tokens=1024,
 messages=[
 {"role": "user", "content": "写一首诗"}
 ]
) as stream:
 for text in stream.text_stream:
 print(text, end="", flush=True)

🔐 授权与账号选择

授权机制

账号选择策略

Token 自动刷新

🏗️ 架构设计

核心模块

📁 项目结构

v2/
├── app.py # FastAPI 主应用
├── db.py # 数据库抽象层 (SQLite/PG/MySQL)
├── replicate.py # Amazon Q 请求复刻
├── auth_flow.py # 设备授权登录
├── claude_types.py # Claude API 类型定义
├── claude_converter.py # Claude 到 Amazon Q 转换
├── claude_parser.py # Event Stream 解析
├── claude_stream.py # Claude SSE 流式处理
├── requirements.txt # Python 依赖
├── .env.example # 环境变量示例
├── .env # 环境变量配置(需自行创建)
├── docker-compose.yml # Docker Compose 配置
├── Dockerfile # Docker 镜像配置
├── data.sqlite3 # SQLite 数据库(自动创建)
├── templates/
│ └── streaming_request.json # Amazon Q 请求模板
├── frontend/
│ └── index.html # Web 控制台
└── scripts/
 ├── account_stats.py # 账号统计脚本
 ├── retry_failed_accounts.py # 重试失败账号脚本
 └── reset_accounts.py # 重置账号脚本

🛠️ 技术栈

🔧 高级配置

环境变量

数据库结构

CREATE TABLE accounts (
 id TEXT PRIMARY KEY, -- UUID
 label TEXT, -- 账号标签
 clientId TEXT, -- OIDC 客户端 ID
 clientSecret TEXT, -- OIDC 客户端密钥
 refreshToken TEXT, -- 刷新令牌
 accessToken TEXT, -- 访问令牌
 other TEXT, -- JSON 格式的额外信息
 last_refresh_time TEXT, -- 最后刷新时间
 last_refresh_status TEXT, -- 最后刷新状态
 created_at TEXT, -- 创建时间
 updated_at TEXT, -- 更新时间
 enabled INTEGER DEFAULT 1, -- 1=启用, 0=禁用
 error_count INTEGER DEFAULT 0, -- 连续错误次数
 success_count INTEGER DEFAULT 0 -- 成功请求次数
);

📝 完整 API 端点列表

管理员认证(需启用 ENABLE_CONSOLE)

账号管理(需启用 ENABLE_CONSOLE,需登录)

设备授权(需启用 ENABLE_CONSOLE,需登录)

OpenAI 兼容

Claude 兼容

其他

🐛 故障排查

工具调用无限循环

401 Unauthorized

Token 刷新失败

🚀 生产环境部署

Uvicorn 多进程模式

# 使用多个 worker 提高并发性能
python -m uvicorn app:app --host 0.0.0.0 --port 8000 --workers 4

Nginx 反向代理

server {
 listen 80;
 server_name your-domain.com;
 location / {
 proxy_pass http://localhost:8000;
 proxy_set_header Host $host;
 proxy_set_header X-Real-IP $remote_addr;
 
 # SSE 支持
 proxy_buffering off;
 proxy_cache off;
 }
}

🔒 安全建议

📄 许可证

🤝 贡献

🙏 致谢