I'm trying to integrate with the Zoho CRM API using Python, and I need to generate an authorization code from a given scope (ZohoCRM.modules.ALL) and then use that authorization code to generate a refresh token.

Currently I am generating code from Zoho developer Console and replacing it manually everytime. I have to generate a new code every time when i am calling API.

• How do I automate the process of obtaining the grant code?
• Is it possible to fully automate this without any manual browser interaction?

Here is the code Than I am using.

CLIENT_ID: str = "12345"
CLIENT_SECRET = "12345"
REDIRECT_URI = "**"
TOKEN_URL = ";
CRM_BASE_URL = ";
ORG_ID = "00000"


# Global variable to store refresh_token
refresh_token = None

class ZohoOAuth:
    def __init__(self):
        # Retrieve sensitive data from environment variables
        self.CLIENT_ID: str = config("CLIENT_ID", cast=str)
        self.CLIENT_SECRET = config("CLIENT_SECRET", cast=str)
        self.REDIRECT_URI = config("REDIRECT_URI", cast=str)
        self.SCOPE = "ZohoCRM.modules.ALL"
        self.token_url = ";

    async def exchange_code_for_tokens(self, code: str):
        # Exchange authorization code for access and refresh tokens
        payload = {
            'grant_type': 'authorization_code',
            'client_id': self.CLIENT_ID,
            'client_secret': self.CLIENT_SECRET,
            'redirect_uri': self.REDIRECT_URI,
            'code': code,
        }
        print(payload)
        try:
            response = requests.post(self.token_url, data=payload)
            response.raise_for_status()  # Will raise an exception for bad status codes
            return response.json()  # Return the access and refresh tokens
        except requests.exceptions.RequestException as e:
            raise HTTPException(status_code=400, detail=f"Error obtaining token: {str(e)}")


@router.get("/generate_token")
async def generate_token(request: Request):
    global refresh_token
    try:
        # Capture authorization code from Zoho redirect
        code = "1000.b500427f0cafcfccc0d456b87503413c.cd5e8d693a143f0054df5bbdb4a706b9"

        if not code:
            raise HTTPException(status_code=400, detail="Authorization code missing")

        # Ensure we await the async method
        tokens = await ZohoOAuth().exchange_code_for_tokens(code)

        if "error" in tokens:
            raise HTTPException(status_code=400, detail=f"Zoho API Error: {tokens['error']}")

        # Store the refresh_token globally after successful exchange
        refresh_token = tokens.get("refresh_token")

        return {
            "access_token": tokens.get("access_token"),
            "refresh_token": refresh_token,
            "scope": tokens.get("scope"),
            "api_domain": tokens.get("api_domain"),
            "token_type": tokens.get("token_type"),
            "expires_in": tokens.get("expires_in")
        }

    except requests.exceptions.RequestException as e:
        print(f"Request error: {str(e)}")  # Log error
        raise HTTPException(status_code=500, detail="Error communicating with Zoho API")

    except Exception as e:
        print(f"Unexpected error: {str(e)}")  # Log error
        raise HTTPException(status_code=500, detail="An unexpected error occurred")


# ✅ Function to get a fresh access token using the global refresh token
def get_access_token():
    global refresh_token

    if not refresh_token:
        raise Exception("Refresh token is missing. Please ensure that you have successfully authenticated.")

    payload = {
        "grant_type": "refresh_token",
        "client_id": CLIENT_ID,
        "client_secret": CLIENT_SECRET,
        "refresh_token": refresh_token,
    }
    response = requests.post(TOKEN_URL, data=payload)
    data = response.json()

    if "access_token" in data:
        return data["access_token"]
    else:
        raise Exception(f"Error fetching access token: {data}")

# Create Zoho Module Endpoint
@router.post("/create_account_module")
async def create_account_modules():
    """Create a new custom module in Zoho CRM."""
    try:
        result = create_account_module()
        return result
    except Exception as e:
        raise HTTPException(status_code=500, detail=str(e))

# ✅ Function to fetch the first available profile ID (mandatory for module creation)
def get_profile_id():
    from src.api.routes.zohocrm import get_access_token
    access_token = get_access_token()
    headers = {"Authorization": f"Zoho-oauthtoken {access_token}", "X-CRM-ORG": ORG_ID}

    response = requests.get(f"{CRM_BASE_URL}/settings/profiles", headers=headers)
    if response.status_code == 200:
        profiles = response.json().get("profiles", [])
        if profiles:
            return profiles[0]["id"]  # Use the first profile ID
        else:
            raise Exception("No profiles found.")
    else:
        raise Exception(f"Error fetching profiles: {response.json()}")



# ✅ Function to create a custom module "DigitalYou"
def create_account_module():
    from src.api.routes.zohocrm import get_access_token
    access_token = get_access_token()  # Ensure this is synchronously called
    profile_id = get_profile_id()      # Ensure this is synchronously called
    print(access_token)
    print(profile_id)

    headers = {
        "Authorization": f"Zoho-oauthtoken {access_token}",
        "X-CRM-ORG": ORG_ID,
        "Content-Type": "application/json"
    }

    module_data = {}

    response = requests.post(f"{CRM_BASE_URL}/settings/modules", json=module_data, headers=headers)

    if response.status_code == 201:
        return response.json()  # ✅ Module created successfully
    else:
        return f"Error: {response.json()}"

After the code used once I am getting error like

Traceback (most recent call last):
  File "/home/roshni/Roshni/AI Avatar/B2B/B2B-DY-Backend/src/api/routes/zohocrm.py", line 62, in generate_token
    raise HTTPException(status_code=400, detail=f"Zoho API Error: {tokens['error']}")
fastapi.exceptions.HTTPException: 400: Zoho API Error: invalid_code

TypeError: 'coroutine' object is not callable

I'm trying to integrate with the Zoho CRM API using Python, and I need to generate an authorization code from a given scope (ZohoCRM.modules.ALL) and then use that authorization code to generate a refresh token.

Currently I am generating code from Zoho developer Console and replacing it manually everytime. I have to generate a new code every time when i am calling API.

• How do I automate the process of obtaining the grant code?
• Is it possible to fully automate this without any manual browser interaction?

Here is the code Than I am using.

CLIENT_ID: str = "12345"
CLIENT_SECRET = "12345"
REDIRECT_URI = "https://accounts.zoho.in**"
TOKEN_URL = "https://accounts.zoho.in/oauth/v2/token"
CRM_BASE_URL = "https://www.zohoapis.in/crm/v7"
ORG_ID = "00000"


# Global variable to store refresh_token
refresh_token = None

class ZohoOAuth:
    def __init__(self):
        # Retrieve sensitive data from environment variables
        self.CLIENT_ID: str = config("CLIENT_ID", cast=str)
        self.CLIENT_SECRET = config("CLIENT_SECRET", cast=str)
        self.REDIRECT_URI = config("REDIRECT_URI", cast=str)
        self.SCOPE = "ZohoCRM.modules.ALL"
        self.token_url = "https://accounts.zoho.in/oauth/v2/token"

    async def exchange_code_for_tokens(self, code: str):
        # Exchange authorization code for access and refresh tokens
        payload = {
            'grant_type': 'authorization_code',
            'client_id': self.CLIENT_ID,
            'client_secret': self.CLIENT_SECRET,
            'redirect_uri': self.REDIRECT_URI,
            'code': code,
        }
        print(payload)
        try:
            response = requests.post(self.token_url, data=payload)
            response.raise_for_status()  # Will raise an exception for bad status codes
            return response.json()  # Return the access and refresh tokens
        except requests.exceptions.RequestException as e:
            raise HTTPException(status_code=400, detail=f"Error obtaining token: {str(e)}")


@router.get("/generate_token")
async def generate_token(request: Request):
    global refresh_token
    try:
        # Capture authorization code from Zoho redirect
        code = "1000.b500427f0cafcfccc0d456b87503413c.cd5e8d693a143f0054df5bbdb4a706b9"

        if not code:
            raise HTTPException(status_code=400, detail="Authorization code missing")

        # Ensure we await the async method
        tokens = await ZohoOAuth().exchange_code_for_tokens(code)

        if "error" in tokens:
            raise HTTPException(status_code=400, detail=f"Zoho API Error: {tokens['error']}")

        # Store the refresh_token globally after successful exchange
        refresh_token = tokens.get("refresh_token")

        return {
            "access_token": tokens.get("access_token"),
            "refresh_token": refresh_token,
            "scope": tokens.get("scope"),
            "api_domain": tokens.get("api_domain"),
            "token_type": tokens.get("token_type"),
            "expires_in": tokens.get("expires_in")
        }

    except requests.exceptions.RequestException as e:
        print(f"Request error: {str(e)}")  # Log error
        raise HTTPException(status_code=500, detail="Error communicating with Zoho API")

    except Exception as e:
        print(f"Unexpected error: {str(e)}")  # Log error
        raise HTTPException(status_code=500, detail="An unexpected error occurred")


# ✅ Function to get a fresh access token using the global refresh token
def get_access_token():
    global refresh_token

    if not refresh_token:
        raise Exception("Refresh token is missing. Please ensure that you have successfully authenticated.")

    payload = {
        "grant_type": "refresh_token",
        "client_id": CLIENT_ID,
        "client_secret": CLIENT_SECRET,
        "refresh_token": refresh_token,
    }
    response = requests.post(TOKEN_URL, data=payload)
    data = response.json()

    if "access_token" in data:
        return data["access_token"]
    else:
        raise Exception(f"Error fetching access token: {data}")

# Create Zoho Module Endpoint
@router.post("/create_account_module")
async def create_account_modules():
    """Create a new custom module in Zoho CRM."""
    try:
        result = create_account_module()
        return result
    except Exception as e:
        raise HTTPException(status_code=500, detail=str(e))

# ✅ Function to fetch the first available profile ID (mandatory for module creation)
def get_profile_id():
    from src.api.routes.zohocrm import get_access_token
    access_token = get_access_token()
    headers = {"Authorization": f"Zoho-oauthtoken {access_token}", "X-CRM-ORG": ORG_ID}

    response = requests.get(f"{CRM_BASE_URL}/settings/profiles", headers=headers)
    if response.status_code == 200:
        profiles = response.json().get("profiles", [])
        if profiles:
            return profiles[0]["id"]  # Use the first profile ID
        else:
            raise Exception("No profiles found.")
    else:
        raise Exception(f"Error fetching profiles: {response.json()}")



# ✅ Function to create a custom module "DigitalYou"
def create_account_module():
    from src.api.routes.zohocrm import get_access_token
    access_token = get_access_token()  # Ensure this is synchronously called
    profile_id = get_profile_id()      # Ensure this is synchronously called
    print(access_token)
    print(profile_id)

    headers = {
        "Authorization": f"Zoho-oauthtoken {access_token}",
        "X-CRM-ORG": ORG_ID,
        "Content-Type": "application/json"
    }

    module_data = {}

    response = requests.post(f"{CRM_BASE_URL}/settings/modules", json=module_data, headers=headers)

    if response.status_code == 201:
        return response.json()  # ✅ Module created successfully
    else:
        return f"Error: {response.json()}"

After the code used once I am getting error like

Traceback (most recent call last):
  File "/home/roshni/Roshni/AI Avatar/B2B/B2B-DY-Backend/src/api/routes/zohocrm.py", line 62, in generate_token
    raise HTTPException(status_code=400, detail=f"Zoho API Error: {tokens['error']}")
fastapi.exceptions.HTTPException: 400: Zoho API Error: invalid_code

TypeError: 'coroutine' object is not callable

• python
• fastapi
• refresh-token
• zoho

Share Improve this question Follow asked Mar 13 at 11:29 Roshni HiraniRoshni Hirani 111 bronze badge

Add a comment  | 

1 Answer 1

Sorted by: Reset to default 0

Is it possible to fully automate this without any manual browser interaction?

No, we cannot do it without browser interaction.

In these scenarios, we need to persist Access and Refresh Token somewhere. This will eliminate the need of authorization_code every time we need 'Access Token'.

The authentication flow on the server side would be,

• Generate a 'code' through the browser flow.
• Generate an access and refresh token with that 'code' and store it.
• Once the access token is expired, generate another access token once more with the refresh token.

Reference: Server-based apps in Zoho API Auth

Tip: If you don't have a place to store the Access token, you can try using Zoho Catalyst's Connectors. It will store the access token in cache and handle all expiry logic for you.