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
1 Answer
Reset to default 0Is 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.