main.py•5.28 kB
# generated by fastapi-codegen:
# filename: openapi.yaml
# timestamp: 2025-06-29T02:22:43+00:00
import argparse
import json
import os
from typing import *
from typing import Optional
from autogen.mcp.mcp_proxy import MCPProxy
from autogen.mcp.mcp_proxy.security import BaseSecurity, UnsuportedSecurityStub
from fastapi import Query
from models import (
Alt,
FieldXgafv,
GoogleAdsHomeservicesLocalservicesV1SearchAccountReportsResponse,
GoogleAdsHomeservicesLocalservicesV1SearchDetailedLeadReportsResponse,
)
app = MCPProxy(
contact={'name': 'Google', 'url': 'https://google.com', 'x-twitter': 'youtube'},
description='',
license={
'name': 'Creative Commons Attribution 3.0',
'url': 'http://creativecommons.org/licenses/by/3.0/',
},
termsOfService='https://developers.google.com/terms/',
title='Local Services API',
version='v1',
servers=[{'url': 'https://localservices.googleapis.com/'}],
)
@app.get(
'/v1/accountReports:search',
description=""" Get account reports containing aggregate account data of all linked GLS accounts. Caller needs to provide their manager customer id and the associated auth credential that allows them read permissions on their linked accounts. """,
tags=['account_report_management', 'lead_report_management'],
security=[
UnsuportedSecurityStub(name="None"),
UnsuportedSecurityStub(name="None"),
],
)
def localservices_account_reports_search(
end_date_day: Optional[int] = Query(None, alias='endDate.day'),
end_date_month: Optional[int] = Query(None, alias='endDate.month'),
end_date_year: Optional[int] = Query(None, alias='endDate.year'),
page_size: Optional[int] = Query(None, alias='pageSize'),
page_token: Optional[str] = Query(None, alias='pageToken'),
query: Optional[str] = None,
start_date_day: Optional[int] = Query(None, alias='startDate.day'),
start_date_month: Optional[int] = Query(None, alias='startDate.month'),
start_date_year: Optional[int] = Query(None, alias='startDate.year'),
field__xgafv: Optional[FieldXgafv] = Query(None, alias='$.xgafv'),
access_token: Optional[str] = None,
alt: Optional[Alt] = None,
callback: Optional[str] = None,
fields: Optional[str] = None,
key: Optional[str] = None,
oauth_token: Optional[str] = None,
pretty_print: Optional[bool] = Query(None, alias='prettyPrint'),
quota_user: Optional[str] = Query(None, alias='quotaUser'),
upload_protocol: Optional[str] = None,
upload_type: Optional[str] = Query(None, alias='uploadType'),
):
raise RuntimeError("Should be patched by MCPProxy and never executed")
@app.get(
'/v1/detailedLeadReports:search',
description=""" Get detailed lead reports containing leads that have been received by all linked GLS accounts. Caller needs to provide their manager customer id and the associated auth credential that allows them read permissions on their linked accounts. """,
tags=['account_report_management', 'lead_report_management'],
security=[
UnsuportedSecurityStub(name="None"),
UnsuportedSecurityStub(name="None"),
],
)
def localservices_detailed_lead_reports_search(
end_date_day: Optional[int] = Query(None, alias='endDate.day'),
end_date_month: Optional[int] = Query(None, alias='endDate.month'),
end_date_year: Optional[int] = Query(None, alias='endDate.year'),
page_size: Optional[int] = Query(None, alias='pageSize'),
page_token: Optional[str] = Query(None, alias='pageToken'),
query: Optional[str] = None,
start_date_day: Optional[int] = Query(None, alias='startDate.day'),
start_date_month: Optional[int] = Query(None, alias='startDate.month'),
start_date_year: Optional[int] = Query(None, alias='startDate.year'),
field__xgafv: Optional[FieldXgafv] = Query(None, alias='$.xgafv'),
access_token: Optional[str] = None,
alt: Optional[Alt] = None,
callback: Optional[str] = None,
fields: Optional[str] = None,
key: Optional[str] = None,
oauth_token: Optional[str] = None,
pretty_print: Optional[bool] = Query(None, alias='prettyPrint'),
quota_user: Optional[str] = Query(None, alias='quotaUser'),
upload_protocol: Optional[str] = None,
upload_type: Optional[str] = Query(None, alias='uploadType'),
):
raise RuntimeError("Should be patched by MCPProxy and never executed")
if __name__ == "__main__":
parser = argparse.ArgumentParser(description="MCP Server")
parser.add_argument(
"transport",
choices=["stdio", "sse", "streamable-http"],
help="Transport mode (stdio, sse or streamable-http)",
)
args = parser.parse_args()
if "CONFIG_PATH" in os.environ:
config_path = os.environ["CONFIG_PATH"]
app.load_configuration(config_path)
if "CONFIG" in os.environ:
config = os.environ["CONFIG"]
app.load_configuration_from_string(config)
if "SECURITY" in os.environ:
security_params = BaseSecurity.parse_security_parameters_from_env(
os.environ,
)
app.set_security_params(security_params)
mcp_settings = json.loads(os.environ.get("MCP_SETTINGS", "{}"))
app.get_mcp(**mcp_settings).run(transport=args.transport)