嗨，我来帮你解决Zoho CRM API授权自动化的问题，顺便修复你代码里的错误～

一、核心问题解答 #

1. 如何自动化获取授权码？ #

有两种靠谱的方案，根据你的场景选：

• 方案1：设备授权流（推荐，无浏览器交互）
  Zoho支持设备授权模式，专门为服务器端自动化场景设计。首次只需要用户手动输入一次验证码，之后全程自动化：

  1. 向Zoho设备授权端点请求，拿到device_code、user_code和验证链接
  2. 用户访问验证链接输入user_code完成授权（仅一次）
  3. 程序轮询令牌端点，直到用户完成授权，自动拿到access_token和refresh_token

• 方案2：无头浏览器自动化
  用Playwright或Selenium模拟浏览器登录Zoho账号，自动点击授权按钮，从跳转URL里提取授权码。适合完全无人干预的场景，但要注意账号安全和验证码拦截问题。

2. 能否完全自动化无浏览器交互？ #

必须可以！用上面的设备授权流就行，首次授权只需要手动操作一次，后续所有令牌刷新都能自动完成，完全不需要浏览器介入。

二、修复你现有代码的错误 #

你遇到的两个错误原因很明确：

1. invalid_code 错误 #

Zoho的授权码只能用一次，你手动复制的代码用过一次就失效了，所以每次调用都会报错。解决办法：

• 改用上面的自动化方案获取授权码
• 保存首次拿到的refresh_token，之后用它来刷新access_token（你的代码里已经有get_access_token函数，只要refresh_token有效就能一直用）

2. TypeError: 'coroutine' object is not callable 错误 #

看你的代码，create_account_module是同步函数，但被异步路由create_account_modules调用，而且里面的get_access_token用了同步的requests。如果要保持异步架构，建议换成httpx异步请求库；如果用同步请求，确保所有函数调用都是同步的，不要混用异步/同步函数导致协程对象未被正确处理。

三、优化后的完整实现示例 #

设备授权流版本（无浏览器交互，首次仅需一次手动验证） #

import requests
import time
from fastapi import FastAPI, HTTPException
from pydantic import BaseSettings

app = FastAPI()

# 用环境变量管理敏感配置，避免硬编码
class Settings(BaseSettings):
    CLIENT_ID: str
    CLIENT_SECRET: str
    ORG_ID: str
    TOKEN_URL: str = "https://accounts.zoho.in/oauth/v2/token"
    DEVICE_AUTH_URL: str = "https://accounts.zoho.in/oauth/v2/device/code"
    CRM_BASE_URL: str = "https://www.zohoapis.in/crm/v7"

settings = Settings()

# 全局存储refresh_token，建议持久化到数据库/配置文件
refresh_token = None

def get_device_code():
    """获取设备授权所需的验证信息"""
    payload = {
        "client_id": settings.CLIENT_ID,
        "scope": "ZohoCRM.modules.ALL"
    }
    response = requests.post(settings.DEVICE_AUTH_URL, data=payload)
    if response.status_code != 200:
        raise HTTPException(status_code=400, detail=f"获取设备码失败: {response.json()}")
    return response.json()

def poll_for_tokens(device_code: str):
    """轮询Zoho服务器，直到用户完成授权"""
    payload = {
        "grant_type": "device_code",
        "client_id": settings.CLIENT_ID,
        "client_secret": settings.CLIENT_SECRET,
        "device_code": device_code
    }
    while True:
        response = requests.post(settings.TOKEN_URL, data=payload)
        data = response.json()
        if response.status_code == 200:
            return data
        elif data.get("error") == "authorization_pending":
            # 按照Zoho返回的间隔时间重试
            time.sleep(data.get("interval", 5))
        else:
            raise HTTPException(status_code=400, detail=f"授权失败: {data}")

@app.get("/init_device_auth")
def init_device_auth():
    """初始化设备授权流程，首次调用"""
    device_data = get_device_code()
    return {
        "message": "请访问下方链接输入验证码完成授权（仅需一次）",
        "verification_url": device_data["verification_url"],
        "user_code": device_data["user_code"],
        "expires_in": device_data["expires_in"]
    }

@app.get("/complete_device_auth")
def complete_device_auth(device_code: str):
    """完成授权，获取并保存refresh_token"""
    global refresh_token
    tokens = poll_for_tokens(device_code)
    refresh_token = tokens.get("refresh_token")
    return {
        "access_token": tokens.get("access_token"),
        "refresh_token": refresh_token,
        "expires_in": tokens.get("expires_in")
    }

def get_access_token():
    """用refresh_token自动刷新access_token"""
    global refresh_token
    if not refresh_token:
        raise HTTPException(status_code=401, detail="未授权，请先调用/init_device_auth完成首次验证")
    
    payload = {
        "grant_type": "refresh_token",
        "client_id": settings.CLIENT_ID,
        "client_secret": settings.CLIENT_SECRET,
        "refresh_token": refresh_token
    }
    response = requests.post(settings.TOKEN_URL, data=payload)
    data = response.json()
    if "access_token" in data:
        return data["access_token"]
    else:
        raise HTTPException(status_code=400, detail=f"刷新令牌失败: {data}")

@app.post("/create_custom_module")
def create_custom_module(module_name: str = "DigitalYou"):
    """创建Zoho CRM自定义模块"""
    try:
        access_token = get_access_token()
        headers = {
            "Authorization": f"Zoho-oauthtoken {access_token}",
            "X-CRM-ORG": settings.ORG_ID
        }

        # 获取必填的profile ID
        profile_response = requests.get(f"{settings.CRM_BASE_URL}/settings/profiles", headers=headers)
        if profile_response.status_code != 200:
            raise HTTPException(status_code=400, detail=f"获取Profile失败: {profile_response.json()}")
        profiles = profile_response.json().get("profiles", [])
        if not profiles:
            raise HTTPException(status_code=400, detail="未找到可用的Profile")
        profile_id = profiles[0]["id"]

        # 模块创建参数
        module_data = {
            "module": {
                "module_name": module_name,
                "plural_label": f"{module_name}s",
                "singular_label": module_name,
                "profile_id": profile_id,
                "is_viewable": True,
                "is_editable": True,
                "is_deletable": True
            }
        }

        # 发送创建请求
        module_response = requests.post(
            f"{settings.CRM_BASE_URL}/settings/modules",
            json=module_data,
            headers={**headers, "Content-Type": "application/json"}
        )
        if module_response.status_code == 201:
            return module_response.json()
        else:
            raise HTTPException(status_code=400, detail=f"创建模块失败: {module_response.json()}")
    except Exception as e:
        raise HTTPException(status_code=500, detail=str(e))

使用步骤 #

1. 首次启动服务后，调用/init_device_auth接口，拿到验证链接和用户验证码
2. 用浏览器打开验证链接，输入验证码完成授权（仅需一次）
3. 调用/complete_device_auth接口，传入init_device_auth返回的device_code，获取并保存refresh_token
4. 之后所有业务接口（如/create_custom_module）都会自动用refresh_token刷新access_token，无需手动操作

四、关键注意事项 #

• refresh_token持久化：建议把refresh_token存到数据库或配置文件里，避免服务重启后丢失
• 权限配置：确保你的Zoho开发者应用已经开启了ZohoCRM.modules.ALL权限
• 错误处理：可以添加更多异常捕获逻辑，比如网络超时、令牌过期自动重试等
• 异步优化：如果你的FastAPI服务是异步架构，推荐用httpx替代requests，实现异步HTTP请求提升性能

备注：内容来源于stack exchange，提问作者Roshni Hirani