AIWriteX 代理合作与接口说明

一、合作说明

AIWriteX 当前已经支持代理模式接入。代理服务器可以为客户端统一下发文本模型与图片模型配置，用户无需手动填写多套模型配置，即可按代理提供的能力直接使用。

适合的合作方向

文本模型代理、图文一体代理、完整能力代理。

当前可形成的服务

默认模型下发、统一中转、套餐控制、字段锁定、日志与配额管理。

当前适合的代理业务模式

仅文本代理：接入门槛最低，最适合快速落地。
文本 + 图片代理：适合提供更完整的图文生成能力。

二、代理版本下载

代理合作方请下载代理版本客户端进行部署与测试。

Windows 代理版 macOS 代理版

三、能力范围

文本模型

可下发文本模型地址
可下发文本模型 Token
可下发文本模型列表
可指定默认模型
可设置默认最大输出长度

图片模型

可下发图片模型地址
可下发图片模型 Token
可下发图片模型列表
可指定默认模型
可设置默认分辨率

微信 API 中继

可作为独立能力下发微信 API 专用中继地址
推荐使用统一授权 Token，由代理服务端按 wechat_relay 能力鉴权、封禁、限额和审计
用于解决微信公众号 IP 白名单频繁变化问题
未返回中继参数时不影响客户端本地配置

字段锁定与能力开关

可控制是否允许用户修改默认代理设置
文本、图片、微信中继三种能力可分别启用、禁用或封禁
可根据套餐返回不同的能力范围

四、请求参数

建议代理服务器提供一个配置接口，例如：POST /v1/client-config。

客户端会以 JSON 方式向代理服务器发送请求。

{
  "agency_id": "D002",
  "license_id": "D002-XXXX-XXXX",
  "machine_id": "4c7d8a2b-xxxx-xxxx",
  "client_version": "1.0.0",
  "contract_version": 2,
  "auth_preference": {
    "header": "Authorization",
    "scheme": "Bearer"
  },
  "capabilities": {
    "text": true,
    "image": true,
    "wechat_relay": true
  }
}

字段	含义	建议用途
agency_id	代理标识	用于识别当前属于哪个代理体系
license_id	用户授权标识	用于查询套餐、状态、额度与权限
machine_id	设备标识	用于风控、设备绑定或审计记录
client_version	客户端版本号	用于版本判断和字段灰度控制
contract_version	代理配置协议版本	当前建议返回 v2 字段，用于统一授权和能力分流
auth_preference	客户端推荐的授权头格式	默认使用 Authorization: Bearer <token>
capabilities	客户端支持的能力范围	用于决定返回文本、图片、微信中继中的哪些项

五、响应结构

代理服务器应返回一个 JSON 对象。

{
  "mode": "agency",
  "ttl_seconds": 300,
  "version": "2026-04-24",
  "authorization": {
    "header": "Authorization",
    "scheme": "Bearer",
    "token": "agt_unified_user_token"
  },
  "locked": {
    "text": true,
    "image": true,
    "wechat_relay": true
  },
  "text": {
    "enabled": true,
    "api_base": "https://proxy.example.com/llm/v1",
    "models": ["gpt-4.1-mini", "gpt-4.1"],
    "model_index": 0,
    "max_tokens": 32768
  },
  "image": {
    "enabled": true,
    "api_base": "https://proxy.example.com/image/v1",
    "models": [
      { "id": "gpt-image-1" },
      { "id": "flux-dev" }
    ],
    "model_index": 0,
    "resolution": "1024x1024"
  },
  "wechat": {
    "relay": {
      "enabled": true,
      "url": "https://relay.example.com",
      "authorization": {
        "header": "Authorization",
        "scheme": "Bearer"
      }
    }
  },
  "policy": {
    "text": { "enabled": true, "quota": "server_enforced" },
    "image": { "enabled": true, "quota": "server_enforced" },
    "wechat_relay": { "enabled": true, "audit": true, "rate_limit": "server_enforced" }
  }
}

建议返回完整、稳定、可直接使用的默认值。文本、图片、微信中继是互相独立的能力：返回有效段时客户端只覆盖该段；未返回某段时客户端保持本地原配置不变。

六、字段与界面含义

文本模型配置

字段	界面含义
enabled	是否显示文本模型的代理配置入口
api_base	文本模型地址
authorization.token	文本模型统一授权 Token；如未单独返回，则使用顶层 authorization.token
models	文本模型下拉列表
model_index	默认选中的文本模型
max_tokens	默认最大输出长度

图片模型配置

字段	界面含义
enabled	是否显示图片模型的代理配置入口
api_base	图片模型地址
authorization.token	图片模型统一授权 Token；如未单独返回，则使用顶层 authorization.token
models	图片模型下拉列表
model_index	默认选中的图片模型
resolution	默认分辨率

微信 API 中继配置

字段	界面含义
wechat.relay.url	微信 API 专用中继地址，例如 https://relay.example.com
wechat.relay.authorization	微信中继统一授权配置；未写 token 时复用顶层 authorization.token

微信中继参数仅代理发行版使用；返回有效中继地址时会启用，未返回则保持客户端本地默认配置不变。中继服务只应转发微信官方 cgi-bin 接口，不应提供通用代理能力。代理服务端应根据用户授权码或统一 Token 的 wechat_relay 能力做启用、封禁、限额和审计。

统一授权与能力分流

authorization.token：统一用户 Token，可同时用于文本、图片、微信中继。
代理服务端根据请求路径、能力类型或内部策略分流到 text、image、wechat_relay。
如果某个能力需要单独 Token，可在该能力段返回 authorization.token 覆盖顶层 Token。

锁定字段与缓存时间

locked：决定当前代理设置是否允许用户修改。
ttl_seconds：决定客户端缓存这份代理配置多久。

七、生效规则

三种能力独立生效

文本、图片、微信中继互不依赖。代理服务器返回某个有效能力段时，客户端只覆盖该能力段；没有返回的能力段保持客户端原有本地配置，不会被清空或回退。

建议的有效性要求

文本：已启用、地址有效、Token 非空、模型列表非空、默认模型索引有效
图片：已启用、地址有效、Token 非空、模型列表非空、默认模型索引有效
微信中继：返回有效 URL 和 authorization.token 时启用，由代理服务端按 wechat_relay 能力鉴权

封禁、限额与审计

客户端只负责携带授权信息和应用有效配置；用户是否可使用 text、image、wechat_relay，应由代理服务端根据授权码、统一 Token、套餐、额度、风控状态进行判断。

八、服务端实现参考

代理服务端建议同时提供配置下发、文本模型代理、图片模型代理和微信 API 中继四类接口。三种能力必须独立判断，用户能用哪个能力由服务端按授权、套餐、额度和风控状态决定。

接口	用途
POST /v1/client-config	AIWriteX 客户端拉取代理配置
/llm/v1/*	文本模型代理，建议兼容 OpenAI Chat Completions 风格
/image/v1/*	图片模型代理，建议兼容 OpenAI Images 风格
/wechat-relay/relay/*	微信 API 专用中继，只转发微信官方 cgi-bin 接口

基础数据结构

USERS = {
    "D002-XXXX-XXXX": {
        "enabled": True,
        "token": "agt_user_token_xxx",
        "capabilities": {
            "text": {"enabled": True, "quota": 1000000},
            "image": {"enabled": True, "quota": 200},
            "wechat_relay": {"enabled": True, "quota": 5000},
        },
    }
}

最小 FastAPI 框架

import hmac
import os
from typing import Any, Dict, Optional

import requests
from fastapi import FastAPI, Header, HTTPException, Request, Response
from pydantic import BaseModel

app = FastAPI(title="AIWriteX Agency Server")

PUBLIC_BASE_URL = os.environ.get("PUBLIC_BASE_URL", "https://proxy.example.com").rstrip("/")
UPSTREAM_TEXT_BASE = os.environ.get("UPSTREAM_TEXT_BASE", "https://api.openai.com/v1").rstrip("/")
UPSTREAM_TEXT_KEY = os.environ.get("UPSTREAM_TEXT_KEY", "")
UPSTREAM_IMAGE_BASE = os.environ.get("UPSTREAM_IMAGE_BASE", "https://api.openai.com/v1").rstrip("/")
UPSTREAM_IMAGE_KEY = os.environ.get("UPSTREAM_IMAGE_KEY", "")
WECHAT_CGI_BASE_URL = "https://api.weixin.qq.com/cgi-bin"

USERS = {
    "D002-XXXX-XXXX": {
        "enabled": True,
        "token": "agt_user_token_xxx",
        "capabilities": {
            "text": {"enabled": True},
            "image": {"enabled": True},
            "wechat_relay": {"enabled": True},
        },
    }
}

HOP_BY_HOP_HEADERS = {
    "connection", "keep-alive", "proxy-authenticate", "proxy-authorization",
    "te", "trailers", "transfer-encoding", "upgrade", "host",
    "content-length", "content-encoding",
}


class ClientConfigRequest(BaseModel):
    agency_id: str = ""
    license_id: str = ""
    machine_id: str = ""
    client_version: str = ""
    contract_version: int = 2
    auth_preference: Dict[str, Any] = {}
    capabilities: Dict[str, bool] = {}


def get_user_by_license(license_id: str) -> Optional[Dict[str, Any]]:
    user = USERS.get(license_id)
    if not user or not user.get("enabled"):
        return None
    return user


def get_token_from_authorization(authorization: Optional[str]) -> str:
    scheme, _, token = (authorization or "").partition(" ")
    return token.strip() if scheme.lower() == "bearer" else ""


def get_user_by_token(token: str) -> Optional[Dict[str, Any]]:
    if not token:
        return None
    for user in USERS.values():
        if user.get("enabled") and hmac.compare_digest(user.get("token", ""), token):
            return user
    return None


def require_capability(authorization: Optional[str], capability: str) -> Dict[str, Any]:
    user = get_user_by_token(get_token_from_authorization(authorization))
    if not user:
        raise HTTPException(status_code=401, detail="授权无效")
    cap = user.get("capabilities", {}).get(capability, {})
    if not cap.get("enabled"):
        raise HTTPException(status_code=403, detail="当前能力未开通或已被禁用")
    return user


def forward_headers(request: Request, upstream_token: str = "") -> Dict[str, str]:
    headers = {
        name: value
        for name, value in request.headers.items()
        if name.lower() not in HOP_BY_HOP_HEADERS and name.lower() != "authorization"
    }
    if upstream_token:
        headers["Authorization"] = f"Bearer {upstream_token}"
    return headers


def response_headers(headers: Dict[str, str]) -> Dict[str, str]:
    return {name: value for name, value in headers.items() if name.lower() not in HOP_BY_HOP_HEADERS}


def proxy_request(request: Request, body: bytes, upstream_url: str, upstream_token: str, timeout: int) -> Response:
    upstream_response = requests.request(
        method=request.method,
        url=upstream_url,
        params=request.url.query,
        data=body,
        headers=forward_headers(request, upstream_token),
        timeout=timeout,
    )
    return Response(
        content=upstream_response.content,
        status_code=upstream_response.status_code,
        headers=response_headers(upstream_response.headers),
        media_type=upstream_response.headers.get("content-type"),
    )


@app.post("/v1/client-config")
def client_config(payload: ClientConfigRequest):
    user = get_user_by_license(payload.license_id)
    if not user:
        return {"mode": "local"}

    token = user["token"]
    caps = user.get("capabilities", {})
    response: Dict[str, Any] = {
        "mode": "agency",
        "ttl_seconds": 300,
        "authorization": {"header": "Authorization", "scheme": "Bearer", "token": token},
        "locked": {},
        "policy": {},
    }

    if caps.get("text", {}).get("enabled"):
        response["locked"]["text"] = True
        response["policy"]["text"] = {"enabled": True, "quota": "server_enforced"}
        response["text"] = {
            "enabled": True,
            "api_base": f"{PUBLIC_BASE_URL}/llm/v1",
            "models": ["gpt-4.1-mini"],
            "model_index": 0,
            "max_tokens": 32768,
        }

    if caps.get("image", {}).get("enabled"):
        response["locked"]["image"] = True
        response["policy"]["image"] = {"enabled": True, "quota": "server_enforced"}
        response["image"] = {
            "enabled": True,
            "api_base": f"{PUBLIC_BASE_URL}/image/v1",
            "models": [{"id": "gpt-image-1"}],
            "model_index": 0,
            "resolution": "1024x1024",
        }

    if caps.get("wechat_relay", {}).get("enabled"):
        response["locked"]["wechat_relay"] = True
        response["policy"]["wechat_relay"] = {"enabled": True, "audit": True, "rate_limit": "server_enforced"}
        response["wechat"] = {
            "relay": {
                "enabled": True,
                "url": f"{PUBLIC_BASE_URL}/wechat-relay",
                "authorization": {"header": "Authorization", "scheme": "Bearer"},
            }
        }

    return response if any(key in response for key in ("text", "image", "wechat")) else {"mode": "local"}


@app.api_route("/llm/v1/{path:path}", methods=["GET", "POST"])
async def llm_proxy(path: str, request: Request, authorization: Optional[str] = Header(default=None)):
    require_capability(authorization, "text")
    body = await request.body()
    return proxy_request(request, body, f"{UPSTREAM_TEXT_BASE}/{path}", UPSTREAM_TEXT_KEY, timeout=120)


@app.api_route("/image/v1/{path:path}", methods=["GET", "POST"])
async def image_proxy(path: str, request: Request, authorization: Optional[str] = Header(default=None)):
    require_capability(authorization, "image")
    body = await request.body()
    return proxy_request(request, body, f"{UPSTREAM_IMAGE_BASE}/{path}", UPSTREAM_IMAGE_KEY, timeout=180)


@app.get("/wechat-relay/relay/meta/egress-ip")
def wechat_relay_egress_ip(authorization: Optional[str] = Header(default=None)):
    require_capability(authorization, "wechat_relay")
    for url in ("https://api.ipify.org", "http://icanhazip.com", "https://ifconfig.me/ip"):
        try:
            resp = requests.get(url, timeout=5)
            resp.raise_for_status()
            if resp.text.strip():
                return {"status": "success", "ip": resp.text.strip()}
        except Exception:
            pass
    raise HTTPException(status_code=502, detail="获取中继出口 IP 失败")


@app.api_route("/wechat-relay/relay/cgi-bin/{wx_path:path}", methods=["GET", "POST"])
async def wechat_relay(wx_path: str, request: Request, authorization: Optional[str] = Header(default=None)):
    require_capability(authorization, "wechat_relay")
    if not wx_path or wx_path.startswith("/") or ".." in wx_path.split("/"):
        raise HTTPException(status_code=400, detail="微信 API 路径无效")

    upstream_response = requests.request(
        method=request.method,
        url=f"{WECHAT_CGI_BASE_URL}/{wx_path}",
        params=request.url.query,
        data=await request.body(),
        headers=forward_headers(request),
        timeout=60,
    )
    return Response(
        content=upstream_response.content,
        status_code=upstream_response.status_code,
        headers=response_headers(upstream_response.headers),
        media_type=upstream_response.headers.get("content-type"),
    )

生产环境必须补齐

用户、套餐、到期时间、统一 Token 必须放到数据库或后台管理系统。
文本、图片、微信中继分别做启用、封禁、限额和审计，不要依赖客户端判断。
不要把上游模型 API Key 返回给客户端；客户端只拿代理服务端签发的用户 Token。
微信中继只允许转发到 https://api.weixin.qq.com/cgi-bin/，不允许任意目标 URL。
微信中继服务器需要固定公网出口 IP，并在微信公众号后台加入白名单。

排查清单

客户端没有代理配置：检查 /v1/client-config 是否返回 mode: agency。
文本不生效：检查 text.enabled、text.api_base、模型列表和顶层 Token。
图片不生效：检查 image.enabled、image.api_base、models[].id 和顶层 Token。
微信中继不生效：检查 wechat.relay.url 是否可访问，以及 Token 是否通过 wechat_relay 能力鉴权。

九、缓存与刷新

自动缓存

客户端会缓存代理服务器返回的配置快照。

自动刷新

缓存到期后，客户端会重新请求代理配置。

手动刷新

只有在某一项已经显示并切换到“自定义”时，界面上才会显示刷新按钮。没有有效代理配置时，不会显示刷新按钮。

建议代理服务器返回稳定的缓存时间与完整配置，避免客户端频繁回退到本地普通模式。

十、合作联系

如果需要进一步了解代理合作、实施方式、套餐能力或商务合作细节，可通过官网页面中的商务联系方式进一步沟通。

推荐先明确：合作模式、预计开放能力、目标用户类型、是否需要文本/图片/微信中继全量支持。
推荐先落地统一授权 Token，再按套餐逐步开放文本、图片、微信中继能力。

AIWriteX 代理合作与接口说明

一、合作说明

推荐的合作切入方式

当前适合的代理业务模式

二、代理版本下载

三、能力范围

文本模型

图片模型

微信 API 中继

字段锁定与能力开关

四、请求参数

五、响应结构

六、字段与界面含义

文本模型配置

图片模型配置

微信 API 中继配置

统一授权与能力分流

锁定字段与缓存时间

七、生效规则

三种能力独立生效

建议的有效性要求

封禁、限额与审计

八、服务端实现参考

基础数据结构

最小 FastAPI 框架

生产环境必须补齐

排查清单

九、缓存与刷新

自动缓存

自动刷新

手动刷新

十、合作联系