main.py•6.82 kB
# generated by fastapi-codegen:
# filename: openapi.yaml
# timestamp: 2025-07-08T17:07:02+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
from fastapi import Query
from models import (
Alt,
FieldXgafv,
ImgColorType,
ImgDominantColor,
ImgSize,
ImgType,
Safe,
Search,
SearchType,
SiteSearchFilter,
)
app = MCPProxy(
contact={'name': 'Google', 'url': 'https://google.com', 'x-twitter': 'youtube'},
description='Searches over a website or collection of websites',
license={
'name': 'Creative Commons Attribution 3.0',
'url': 'http://creativecommons.org/licenses/by/3.0/',
},
termsOfService='https://developers.google.com/terms/',
title='Custom Search API',
version='v1',
servers=[{'url': 'https://customsearch.googleapis.com/'}],
)
@app.get(
'/customsearch/v1',
description=""" Returns metadata about the search performed, metadata about the engine used for the search, and the search results. """,
tags=['cse_search_operations'],
)
def search_cse_list(
c2coff: Optional[str] = None,
cr: Optional[str] = None,
cx: Optional[str] = None,
date_restrict: Optional[str] = Query(None, alias='dateRestrict'),
exact_terms: Optional[str] = Query(None, alias='exactTerms'),
exclude_terms: Optional[str] = Query(None, alias='excludeTerms'),
file_type: Optional[str] = Query(None, alias='fileType'),
filter: Optional[str] = None,
gl: Optional[str] = None,
googlehost: Optional[str] = None,
high_range: Optional[str] = Query(None, alias='highRange'),
hl: Optional[str] = None,
hq: Optional[str] = None,
img_color_type: Optional[ImgColorType] = Query(None, alias='imgColorType'),
img_dominant_color: Optional[ImgDominantColor] = Query(
None, alias='imgDominantColor'
),
img_size: Optional[ImgSize] = Query(None, alias='imgSize'),
img_type: Optional[ImgType] = Query(None, alias='imgType'),
link_site: Optional[str] = Query(None, alias='linkSite'),
low_range: Optional[str] = Query(None, alias='lowRange'),
lr: Optional[str] = None,
num: Optional[int] = None,
or_terms: Optional[str] = Query(None, alias='orTerms'),
q: Optional[str] = None,
related_site: Optional[str] = Query(None, alias='relatedSite'),
rights: Optional[str] = None,
safe: Optional[Safe] = None,
search_type: Optional[SearchType] = Query(None, alias='searchType'),
site_search: Optional[str] = Query(None, alias='siteSearch'),
site_search_filter: Optional[SiteSearchFilter] = Query(
None, alias='siteSearchFilter'
),
sort: Optional[str] = None,
start: Optional[int] = None,
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(
'/customsearch/v1/siterestrict',
description=""" Returns metadata about the search performed, metadata about the engine used for the search, and the search results. Uses a small set of url patterns. """,
tags=['cse_search_operations'],
)
def search_cse_siterestrict_list(
c2coff: Optional[str] = None,
cr: Optional[str] = None,
cx: Optional[str] = None,
date_restrict: Optional[str] = Query(None, alias='dateRestrict'),
exact_terms: Optional[str] = Query(None, alias='exactTerms'),
exclude_terms: Optional[str] = Query(None, alias='excludeTerms'),
file_type: Optional[str] = Query(None, alias='fileType'),
filter: Optional[str] = None,
gl: Optional[str] = None,
googlehost: Optional[str] = None,
high_range: Optional[str] = Query(None, alias='highRange'),
hl: Optional[str] = None,
hq: Optional[str] = None,
img_color_type: Optional[ImgColorType] = Query(None, alias='imgColorType'),
img_dominant_color: Optional[ImgDominantColor] = Query(
None, alias='imgDominantColor'
),
img_size: Optional[ImgSize] = Query(None, alias='imgSize'),
img_type: Optional[ImgType] = Query(None, alias='imgType'),
link_site: Optional[str] = Query(None, alias='linkSite'),
low_range: Optional[str] = Query(None, alias='lowRange'),
lr: Optional[str] = None,
num: Optional[int] = None,
or_terms: Optional[str] = Query(None, alias='orTerms'),
q: Optional[str] = None,
related_site: Optional[str] = Query(None, alias='relatedSite'),
rights: Optional[str] = None,
safe: Optional[Safe] = None,
search_type: Optional[SearchType] = Query(None, alias='searchType'),
site_search: Optional[str] = Query(None, alias='siteSearch'),
site_search_filter: Optional[SiteSearchFilter] = Query(
None, alias='siteSearchFilter'
),
sort: Optional[str] = None,
start: Optional[int] = None,
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)