main.py•26.1 kB
# generated by fastapi-codegen:
# filename: openapi.yaml
# timestamp: 2025-06-29T01:47:46+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,
CancelOperationRequest,
ConnectionProfile,
DiscoverConnectionProfileRequest,
DiscoverConnectionProfileResponse,
Empty,
FetchStaticIpsResponse,
FieldXgafv,
ListConnectionProfilesResponse,
ListLocationsResponse,
ListOperationsResponse,
ListPrivateConnectionsResponse,
ListRoutesResponse,
ListStreamObjectsResponse,
ListStreamsResponse,
LookupStreamObjectRequest,
Operation,
PrivateConnection,
Route,
StartBackfillJobRequest,
StartBackfillJobResponse,
StopBackfillJobRequest,
StopBackfillJobResponse,
Stream,
StreamObject,
)
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='Datastream API',
version='v1',
servers=[{'url': 'https://datastream.googleapis.com/'}],
)
@app.delete(
'/v1/{name}',
description=""" Use this method to delete a stream. """,
tags=['datastream_management', 'datastream_operations'],
security=[
UnsuportedSecurityStub(name="None"),
UnsuportedSecurityStub(name="None"),
],
)
def datastream_projects_locations_streams_delete(
name: str,
request_id: Optional[str] = Query(None, alias='requestId'),
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/{name}',
description=""" Use this method to get details about a stream object. """,
tags=[
'datastream_management',
'datastream_operations',
'datastream_objects',
'datastream_connection_profiles',
'datastream_private_connections',
],
security=[
UnsuportedSecurityStub(name="None"),
UnsuportedSecurityStub(name="None"),
],
)
def datastream_projects_locations_streams_objects_get(
name: str,
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.patch(
'/v1/{name}',
description=""" Use this method to update the configuration of a stream. """,
tags=['datastream_management', 'datastream_operations'],
security=[
UnsuportedSecurityStub(name="None"),
UnsuportedSecurityStub(name="None"),
],
)
def datastream_projects_locations_streams_patch(
name: str,
force: Optional[bool] = None,
request_id: Optional[str] = Query(None, alias='requestId'),
update_mask: Optional[str] = Query(None, alias='updateMask'),
validate_only: Optional[bool] = Query(None, alias='validateOnly'),
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'),
body: Stream = None,
):
raise RuntimeError("Should be patched by MCPProxy and never executed")
@app.get(
'/v1/{name}/locations',
description=""" Lists information about the supported locations for this service. """,
tags=['datastream_management', 'datastream_operations'],
security=[
UnsuportedSecurityStub(name="None"),
UnsuportedSecurityStub(name="None"),
],
)
def datastream_projects_locations_list(
name: str,
filter: Optional[str] = None,
page_size: Optional[int] = Query(None, alias='pageSize'),
page_token: Optional[str] = Query(None, alias='pageToken'),
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/{name}/operations',
description=""" Lists operations that match the specified filter in the request. If the server doesn't support this method, it returns `UNIMPLEMENTED`. """,
tags=[
'datastream_management',
'datastream_operations',
'datastream_connection_profiles',
],
security=[
UnsuportedSecurityStub(name="None"),
UnsuportedSecurityStub(name="None"),
],
)
def datastream_projects_locations_operations_list(
name: str,
filter: Optional[str] = None,
page_size: Optional[int] = Query(None, alias='pageSize'),
page_token: Optional[str] = Query(None, alias='pageToken'),
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.post(
'/v1/{name}:cancel',
description=""" Starts asynchronous cancellation on a long-running operation. The server makes a best effort to cancel the operation, but success is not guaranteed. If the server doesn't support this method, it returns `google.rpc.Code.UNIMPLEMENTED`. Clients can use Operations.GetOperation or other methods to check whether the cancellation succeeded or whether the operation completed despite cancellation. On successful cancellation, the operation is not deleted; instead, it becomes an operation with an Operation.error value with a google.rpc.Status.code of 1, corresponding to `Code.CANCELLED`. """,
tags=['datastream_operations'],
security=[
UnsuportedSecurityStub(name="None"),
UnsuportedSecurityStub(name="None"),
],
)
def datastream_projects_locations_operations_cancel(
name: str,
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'),
body: CancelOperationRequest = None,
):
raise RuntimeError("Should be patched by MCPProxy and never executed")
@app.get(
'/v1/{name}:fetchStaticIps',
description=""" The FetchStaticIps API call exposes the static IP addresses used by Datastream. """,
tags=['datastream_management'],
security=[
UnsuportedSecurityStub(name="None"),
UnsuportedSecurityStub(name="None"),
],
)
def datastream_projects_locations_fetch_static_ips(
name: str,
page_size: Optional[int] = Query(None, alias='pageSize'),
page_token: Optional[str] = Query(None, alias='pageToken'),
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.post(
'/v1/{object}:startBackfillJob',
description=""" Use this method to start a backfill job for the specified stream object. """,
tags=['datastream_operations'],
security=[
UnsuportedSecurityStub(name="None"),
UnsuportedSecurityStub(name="None"),
],
)
def datastream_projects_locations_streams_objects_start_backfill_job(
object: str,
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'),
body: StartBackfillJobRequest = None,
):
raise RuntimeError("Should be patched by MCPProxy and never executed")
@app.post(
'/v1/{object}:stopBackfillJob',
description=""" Use this method to stop a backfill job for the specified stream object. """,
tags=['datastream_operations'],
security=[
UnsuportedSecurityStub(name="None"),
UnsuportedSecurityStub(name="None"),
],
)
def datastream_projects_locations_streams_objects_stop_backfill_job(
object: str,
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'),
body: StopBackfillJobRequest = None,
):
raise RuntimeError("Should be patched by MCPProxy and never executed")
@app.get(
'/v1/{parent}/connectionProfiles',
description=""" Use this method to list connection profiles created in a project and location. """,
tags=['datastream_management', 'datastream_connection_profiles'],
security=[
UnsuportedSecurityStub(name="None"),
UnsuportedSecurityStub(name="None"),
],
)
def datastream_projects_locations_connection_profiles_list(
parent: str,
filter: Optional[str] = None,
order_by: Optional[str] = Query(None, alias='orderBy'),
page_size: Optional[int] = Query(None, alias='pageSize'),
page_token: Optional[str] = Query(None, alias='pageToken'),
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.post(
'/v1/{parent}/connectionProfiles',
description=""" Use this method to create a connection profile in a project and location. """,
tags=['datastream_connection_profiles'],
security=[
UnsuportedSecurityStub(name="None"),
UnsuportedSecurityStub(name="None"),
],
)
def datastream_projects_locations_connection_profiles_create(
parent: str,
connection_profile_id: Optional[str] = Query(None, alias='connectionProfileId'),
force: Optional[bool] = None,
request_id: Optional[str] = Query(None, alias='requestId'),
validate_only: Optional[bool] = Query(None, alias='validateOnly'),
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'),
body: ConnectionProfile = None,
):
raise RuntimeError("Should be patched by MCPProxy and never executed")
@app.post(
'/v1/{parent}/connectionProfiles:discover',
description=""" Use this method to discover a connection profile. The discover API call exposes the data objects and metadata belonging to the profile. Typically, a request returns children data objects of a parent data object that's optionally supplied in the request. """,
tags=['datastream_connection_profiles'],
security=[
UnsuportedSecurityStub(name="None"),
UnsuportedSecurityStub(name="None"),
],
)
def datastream_projects_locations_connection_profiles_discover(
parent: str,
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'),
body: DiscoverConnectionProfileRequest = None,
):
raise RuntimeError("Should be patched by MCPProxy and never executed")
@app.get(
'/v1/{parent}/objects',
description=""" Use this method to list the objects of a specific stream. """,
tags=[
'datastream_management',
'datastream_operations',
'datastream_objects',
'datastream_connection_profiles',
'datastream_private_connections',
],
security=[
UnsuportedSecurityStub(name="None"),
UnsuportedSecurityStub(name="None"),
],
)
def datastream_projects_locations_streams_objects_list(
parent: str,
page_size: Optional[int] = Query(None, alias='pageSize'),
page_token: Optional[str] = Query(None, alias='pageToken'),
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.post(
'/v1/{parent}/objects:lookup',
description=""" Use this method to look up a stream object by its source object identifier. """,
tags=['datastream_objects'],
security=[
UnsuportedSecurityStub(name="None"),
UnsuportedSecurityStub(name="None"),
],
)
def datastream_projects_locations_streams_objects_lookup(
parent: str,
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'),
body: LookupStreamObjectRequest = None,
):
raise RuntimeError("Should be patched by MCPProxy and never executed")
@app.get(
'/v1/{parent}/privateConnections',
description=""" Use this method to list private connectivity configurations in a project and location. """,
tags=['datastream_management', 'datastream_connection_profiles'],
security=[
UnsuportedSecurityStub(name="None"),
UnsuportedSecurityStub(name="None"),
],
)
def datastream_projects_locations_private_connections_list(
parent: str,
filter: Optional[str] = None,
order_by: Optional[str] = Query(None, alias='orderBy'),
page_size: Optional[int] = Query(None, alias='pageSize'),
page_token: Optional[str] = Query(None, alias='pageToken'),
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.post(
'/v1/{parent}/privateConnections',
description=""" Use this method to create a private connectivity configuration. """,
tags=['datastream_private_connections'],
security=[
UnsuportedSecurityStub(name="None"),
UnsuportedSecurityStub(name="None"),
],
)
def datastream_projects_locations_private_connections_create(
parent: str,
force: Optional[bool] = None,
private_connection_id: Optional[str] = Query(None, alias='privateConnectionId'),
request_id: Optional[str] = Query(None, alias='requestId'),
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'),
body: PrivateConnection = None,
):
raise RuntimeError("Should be patched by MCPProxy and never executed")
@app.get(
'/v1/{parent}/routes',
description=""" Use this method to list routes created for a private connectivity configuration in a project and location. """,
tags=['datastream_management', 'datastream_objects'],
security=[
UnsuportedSecurityStub(name="None"),
UnsuportedSecurityStub(name="None"),
],
)
def datastream_projects_locations_private_connections_routes_list(
parent: str,
filter: Optional[str] = None,
order_by: Optional[str] = Query(None, alias='orderBy'),
page_size: Optional[int] = Query(None, alias='pageSize'),
page_token: Optional[str] = Query(None, alias='pageToken'),
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.post(
'/v1/{parent}/routes',
description=""" Use this method to create a route for a private connectivity configuration in a project and location. """,
tags=[
'datastream_management',
'datastream_operations',
'datastream_objects',
'datastream_connection_profiles',
'datastream_private_connections',
],
security=[
UnsuportedSecurityStub(name="None"),
UnsuportedSecurityStub(name="None"),
],
)
def datastream_projects_locations_private_connections_routes_create(
parent: str,
request_id: Optional[str] = Query(None, alias='requestId'),
route_id: Optional[str] = Query(None, alias='routeId'),
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'),
body: Route = None,
):
raise RuntimeError("Should be patched by MCPProxy and never executed")
@app.get(
'/v1/{parent}/streams',
description=""" Use this method to list streams in a project and location. """,
tags=[
'datastream_management',
'datastream_operations',
'datastream_connection_profiles',
],
security=[
UnsuportedSecurityStub(name="None"),
UnsuportedSecurityStub(name="None"),
],
)
def datastream_projects_locations_streams_list(
parent: str,
filter: Optional[str] = None,
order_by: Optional[str] = Query(None, alias='orderBy'),
page_size: Optional[int] = Query(None, alias='pageSize'),
page_token: Optional[str] = Query(None, alias='pageToken'),
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.post(
'/v1/{parent}/streams',
description=""" Use this method to create a stream. """,
tags=['datastream_management', 'datastream_operations'],
security=[
UnsuportedSecurityStub(name="None"),
UnsuportedSecurityStub(name="None"),
],
)
def datastream_projects_locations_streams_create(
parent: str,
force: Optional[bool] = None,
request_id: Optional[str] = Query(None, alias='requestId'),
stream_id: Optional[str] = Query(None, alias='streamId'),
validate_only: Optional[bool] = Query(None, alias='validateOnly'),
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'),
body: Stream = None,
):
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)