main.py•2.72 kB
# generated by fastapi-codegen:
# filename: openapi.yaml
# timestamp: 2025-07-08T17:49:41+00:00
import argparse
import json
import os
from typing import *
from autogen.mcp.mcp_proxy import MCPProxy
from autogen.mcp.mcp_proxy.security import BaseSecurity
from models import InputSchema, RunsResponseSchema
app = MCPProxy(
description="Extract all reviews of Google Maps places using place URLs. Get review text, published date, response from owner, review URL, and reviewer's details. Download scraped data, run the scraper via API, schedule and monitor runs or integrate with other tools.",
title='Google Maps Reviews Scraper',
version='0.0',
servers=[{'url': 'https://api.apify.com/v2'}],
)
@app.post(
'/acts/compass~Google-Maps-Reviews-Scraper/run-sync', tags=['actor_execution']
)
def run_sync_compass__google__maps__reviews__scraper(
token: str, body: InputSchema = ...
):
"""
Executes an Actor, waits for completion, and returns the OUTPUT from Key-value store in response.
"""
raise RuntimeError("Should be patched by MCPProxy and never executed")
@app.post(
'/acts/compass~Google-Maps-Reviews-Scraper/run-sync-get-dataset-items',
tags=['actor_execution'],
)
def execute_actor_and_return_dataset_items(token: str, body: InputSchema = ...):
"""
Executes an Actor, waits for its completion, and returns Actor's dataset items in response.
"""
raise RuntimeError("Should be patched by MCPProxy and never executed")
@app.post('/acts/compass~Google-Maps-Reviews-Scraper/runs', tags=['actor_execution'])
def runs_sync_compass__google__maps__reviews__scraper(
token: str, body: InputSchema = ...
):
"""
Executes an Actor and returns information about the initiated run in response.
"""
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)