Hello, I successfully connected my custom Google Ads connector, but the search results are not good. I understand that both search and fetch are mandatory for building MCP connectors, so I developed the code accordingly and connected it. However, I’m not getting the expected results—only dummy data is returned by ChatGPT, even though search and fetch work correctly locally and on Replit.
import os
import logging
from typing import Dict, Any, List
from fastmcp.server import FastMCP
from pydantic import BaseModel
logging.basicConfig(level=logging.INFO)
public_url = “https://gads-server-replit2844.replit.app”
GOOGLE_ADS_DEVELOPER_TOKEN = os.environ.get(“GOOGLE_ADS_DEVELOPER_TOKEN”)
GOOGLE_ADS_CLIENT_ID = os.environ.get(“GOOGLE_ADS_CLIENT_ID”)
GOOGLE_ADS_CLIENT_SECRET = os.environ.get(“GOOGLE_ADS_CLIENT_SECRET”)
GOOGLE_ADS_REFRESH_TOKEN = os.environ.get(“GOOGLE_ADS_REFRESH_TOKEN”)
GOOGLE_ADS_CUSTOMER_ID = os.environ.get(“GOOGLE_ADS_CUSTOMER_ID”)
GOOGLE_ADS_LOGIN_CUSTOMER_ID = os.environ.get(“GOOGLE_ADS_LOGIN_CUSTOMER_ID”) # MCC ID
app = FastMCP(
name=“Google Ads Connector”,
instructions=“Connects to Google Ads API to search campaigns, keywords, and fetch detailed performance data.”,
)
def get_google_ads_client():
“”“Initialize and return Google Ads API client”“”
try:
from google.ads.googleads.client import GoogleAdsClient
config = {
"developer_token": GOOGLE_ADS_DEVELOPER_TOKEN,
"client_id": GOOGLE_ADS_CLIENT_ID,
"client_secret": GOOGLE_ADS_CLIENT_SECRET,
"refresh_token": GOOGLE_ADS_REFRESH_TOKEN,
"use_proto_plus": True, # Required as of v14.0.0
}
# Add login customer ID if provided (for MCC access)
if GOOGLE_ADS_LOGIN_CUSTOMER_ID:
config["login_customer_id"] = GOOGLE_ADS_LOGIN_CUSTOMER_ID
# Validate required credentials
missing_creds = [k for k, v in config.items() if not v and k not in ["use_proto_plus", "login_customer_id"]]
if missing_creds:
raise ValueError(f"Missing Google Ads credentials: {missing_creds}")
return GoogleAdsClient.load_from_dict(config)
except ImportError:
raise ImportError(
"Google Ads library not installed. Run: pip install google-ads"
)
except Exception as e:
logging.error(f"Failed to initialize Google Ads client: {e}")
raise
def build_campaign_query(search_terms: str = None) → str:
“”“Build GAQL query for campaign search”“”
base_query = “”"
SELECT
campaign.id,
campaign.name,
campaign.status
FROM campaign
LIMIT 10
“”"
return base_query
def build_campaign_detail_query(campaign_id: str) → str:
“”“Build GAQL query for specific campaign details”“”
return f"“”
SELECT
campaign.id,
campaign.name,
campaign.status,
campaign.advertising_channel_type,
campaign.start_date,
campaign.end_date,
campaign_budget.amount_micros,
metrics.impressions,
metrics.clicks,
metrics.ctr,
metrics.cost_micros,
metrics.conversions,
metrics.average_cpc,
metrics.average_cpm
FROM campaign
WHERE campaign.id = {campaign_id}
“”"
def micros_to_currency(micros: int) → float:
“”“Convert micros to currency format”“”
return micros / 1_000_000 if micros else 0
def format_campaign_data(row) → Dict[str, Any]:
“”“Format campaign data from API response”“”
campaign = row.campaign
return {
"id": str(campaign.id),
"name": campaign.name,
"status": campaign.status.name
}
def format_campaign_details(row) → Dict[str, Any]:
“”“Format detailed campaign data from API response”“”
campaign = row.campaign
metrics = row.metrics
budget = getattr(row, ‘campaign_budget’, None)
details = format_campaign_data(row)
# Add detailed information
details.update({
"start_date": campaign.start_date if hasattr(campaign, 'start_date') else None,
"end_date": campaign.end_date if hasattr(campaign, 'end_date') else None,
"budget": {
"daily_budget": f"${micros_to_currency(budget.amount_micros):.2f}" if budget else "Not available"
},
"detailed_metrics": {
"average_cpc": f"${micros_to_currency(metrics.average_cpc):.2f}" if metrics.average_cpc else "$0.00",
"average_cpm": f"${micros_to_currency(metrics.average_cpm):.2f}" if metrics.average_cpm else "$0.00"
}
})
return details
@app.tool()
async def search(query: str) → Dict[str, Any]:
“”"
Search Google Ads campaigns, keywords, or performance data.
Args:
query: Search term for Google Ads data (e.g., "campaign performance", "keyword metrics", "brand campaigns")
Returns:
Dictionary containing search results with campaign data
"""
try:
# Check if Google Ads credentials are configured
if not all([GOOGLE_ADS_DEVELOPER_TOKEN, GOOGLE_ADS_CLIENT_ID,
GOOGLE_ADS_CLIENT_SECRET, GOOGLE_ADS_REFRESH_TOKEN,
GOOGLE_ADS_CUSTOMER_ID]):
return {
"error": "Google Ads API credentials not configured",
"message": "Please set up your Google Ads API credentials in environment variables",
"required_variables": [
"GOOGLE_ADS_DEVELOPER_TOKEN",
"GOOGLE_ADS_CLIENT_ID",
"GOOGLE_ADS_CLIENT_SECRET",
"GOOGLE_ADS_REFRESH_TOKEN",
"GOOGLE_ADS_CUSTOMER_ID",
"GOOGLE_ADS_LOGIN_CUSTOMER_ID (for MCC access)"
]
}
try:
# Initialize Google Ads client
client = get_google_ads_client()
service = client.get_service("GoogleAdsService")
# Build and execute query
gaql_query = build_campaign_query(query)
response = service.search(
customer_id=GOOGLE_ADS_CUSTOMER_ID,
query=gaql_query
)
# Process results
campaigns = []
for row in response:
campaign_data = format_campaign_data(row)
campaigns.append(campaign_data)
# Optional: If search terms provided, we could filter here
# But for now, let's return all campaigns
return {
"query": query,
"total_results": len(campaigns),
"campaigns": campaigns,
"customer_id": GOOGLE_ADS_CUSTOMER_ID
}
except Exception as api_error:
logging.error(f"Google Ads API error: {api_error}")
return {
"error": "Google Ads API request failed",
"message": str(api_error),
"query": query
}
except Exception as e:
logging.error(f"Search error: {e}")
return {
"error": "Search request failed",
"message": str(e),
"query": query
}
@app.tool()
async def fetch(id: str) → Dict[str, Any]:
“”"
Fetch detailed information about a specific Google Ads campaign.
Args:
id: The campaign ID to fetch detailed information for
Returns:
Dictionary containing detailed campaign information
"""
try:
# Check if Google Ads credentials are configured
if not all([GOOGLE_ADS_DEVELOPER_TOKEN, GOOGLE_ADS_CLIENT_ID,
GOOGLE_ADS_CLIENT_SECRET, GOOGLE_ADS_REFRESH_TOKEN,
GOOGLE_ADS_CUSTOMER_ID]):
return {
"error": "Google Ads API credentials not configured",
"message": "Please set up your Google Ads API credentials in environment variables"
}
try:
# Initialize Google Ads client
client = get_google_ads_client()
service = client.get_service("GoogleAdsService")
# Build and execute query for specific campaign
gaql_query = build_campaign_detail_query(id)
response = service.search(
customer_id=GOOGLE_ADS_CUSTOMER_ID,
query=gaql_query
)
# Process result
for row in response:
return format_campaign_details(row)
# If no results found
return {
"error": "Campaign not found",
"message": f"No campaign found with ID: {id}",
"campaign_id": id
}
except Exception as api_error:
logging.error(f"Google Ads API error: {api_error}")
return {
"error": "Google Ads API request failed",
"message": str(api_error),
"campaign_id": id
}
except Exception as e:
logging.error(f"Fetch error: {e}")
return {
"error": "Fetch request failed",
"message": str(e),
"campaign_id": id
}
@app.tool()
async def health_check() → Dict[str, Any]:
“”"
Check the health and configuration status of the Google Ads connector.
Returns:
Dictionary containing health status and configuration info
"""
config_status = {
"GOOGLE_ADS_DEVELOPER_TOKEN": "✓ Set" if GOOGLE_ADS_DEVELOPER_TOKEN else "✗ Missing",
"GOOGLE_ADS_CLIENT_ID": "✓ Set" if GOOGLE_ADS_CLIENT_ID else "✗ Missing",
"GOOGLE_ADS_CLIENT_SECRET": "✓ Set" if GOOGLE_ADS_CLIENT_SECRET else "✗ Missing",
"GOOGLE_ADS_REFRESH_TOKEN": "✓ Set" if GOOGLE_ADS_REFRESH_TOKEN else "✗ Missing",
"GOOGLE_ADS_CUSTOMER_ID": "✓ Set" if GOOGLE_ADS_CUSTOMER_ID else "✗ Missing",
"GOOGLE_ADS_LOGIN_CUSTOMER_ID": "✓ Set" if GOOGLE_ADS_LOGIN_CUSTOMER_ID else "✗ Missing (needed for MCC)"
}
try:
# Try to import Google Ads library
from google.ads.googleads.client import GoogleAdsClient
library_status = "✓ Installed"
except ImportError:
library_status = "✗ Not installed - Run: pip install google-ads"
return {
"status": "healthy" if all([GOOGLE_ADS_DEVELOPER_TOKEN, GOOGLE_ADS_CLIENT_ID,
GOOGLE_ADS_CLIENT_SECRET, GOOGLE_ADS_REFRESH_TOKEN,
GOOGLE_ADS_CUSTOMER_ID]) else "configuration_needed",
"google_ads_library": library_status,
"configuration": config_status,
"server_url": public_url
}
if name == “main”:
app.run(transport=“streamable-http”, host=“0.0.0.0”, port=8000, path=“/mcp”)