Dataproc Metastore API MCP Server

# generated by fastapi-codegen: # filename: openapi.yaml # timestamp: 2025-06-29T02:25:26+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, AlterMetadataResourceLocationRequest, Backup, CancelOperationRequest, Empty, ExportMetadataRequest, Federation, FieldXgafv, ListBackupsResponse, ListFederationsResponse, ListLocationsResponse, ListMetadataImportsResponse, ListOperationsResponse, ListServicesResponse, MetadataImport, MoveTableToDatabaseRequest, Operation, Policy, QueryMetadataRequest, RemoveIamPolicyRequest, RemoveIamPolicyResponse, RestoreServiceRequest, Service, SetIamPolicyRequest, TestIamPermissionsRequest, TestIamPermissionsResponse, ) app = MCPProxy( contact={'name': 'Google', 'url': 'https://google.com', 'x-twitter': 'youtube'}, description='The Dataproc Metastore API is used to manage the lifecycle and configuration of metastore services.', license={ 'name': 'Creative Commons Attribution 3.0', 'url': 'http://creativecommons.org/licenses/by/3.0/', }, termsOfService='https://developers.google.com/terms/', title='Dataproc Metastore API', version='v1beta', servers=[{'url': 'https://metastore.googleapis.com/'}], ) @app.delete( '/v1beta/{name}', description=""" Deletes a single backup. """, tags=['metastore_operations_handling'], security=[ UnsuportedSecurityStub(name="None"), UnsuportedSecurityStub(name="None"), ], ) def metastore_projects_locations_services_backups_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( '/v1beta/{name}', description=""" Gets details of a single import. """, tags=['metastore_service_configuration'], security=[ UnsuportedSecurityStub(name="None"), UnsuportedSecurityStub(name="None"), ], ) def metastore_projects_locations_services_metadata_imports_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( '/v1beta/{name}', description=""" Updates a single import. Only the description field of MetadataImport is supported to be updated. """, tags=['metastore_metadata_import_handling'], security=[ UnsuportedSecurityStub(name="None"), UnsuportedSecurityStub(name="None"), ], ) def metastore_projects_locations_services_metadata_imports_patch( name: str, request_id: Optional[str] = Query(None, alias='requestId'), update_mask: Optional[str] = Query(None, alias='updateMask'), 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: MetadataImport = None, ): raise RuntimeError("Should be patched by MCPProxy and never executed") @app.get( '/v1beta/{name}/locations', description=""" Lists information about the supported locations for this service. """, tags=['metastore_service_configuration'], security=[ UnsuportedSecurityStub(name="None"), UnsuportedSecurityStub(name="None"), ], ) def metastore_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( '/v1beta/{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=['metastore_service_configuration'], security=[ UnsuportedSecurityStub(name="None"), UnsuportedSecurityStub(name="None"), ], ) def metastore_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( '/v1beta/{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=['metastore_operations_handling'], security=[ UnsuportedSecurityStub(name="None"), UnsuportedSecurityStub(name="None"), ], ) def metastore_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( '/v1beta/{parent}/backups', description=""" Lists backups in a service. """, tags=['metastore_operations_handling', 'metastore_service_configuration'], security=[ UnsuportedSecurityStub(name="None"), UnsuportedSecurityStub(name="None"), ], ) def metastore_projects_locations_services_backups_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( '/v1beta/{parent}/backups', description=""" Creates a new backup in a given project and location. """, tags=['metastore_backup_handling'], security=[ UnsuportedSecurityStub(name="None"), UnsuportedSecurityStub(name="None"), ], ) def metastore_projects_locations_services_backups_create( parent: str, backup_id: Optional[str] = Query(None, alias='backupId'), 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: Backup = None, ): raise RuntimeError("Should be patched by MCPProxy and never executed") @app.get( '/v1beta/{parent}/federations', description=""" Lists federations in a project and location. """, tags=['metastore_service_configuration'], security=[ UnsuportedSecurityStub(name="None"), UnsuportedSecurityStub(name="None"), ], ) def metastore_projects_locations_federations_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( '/v1beta/{parent}/federations', description=""" Creates a metastore federation in a project and location. """, tags=['metastore_federation_management', 'metastore_service_configuration'], security=[ UnsuportedSecurityStub(name="None"), UnsuportedSecurityStub(name="None"), ], ) def metastore_projects_locations_federations_create( parent: str, federation_id: Optional[str] = Query(None, alias='federationId'), 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: Federation = None, ): raise RuntimeError("Should be patched by MCPProxy and never executed") @app.get( '/v1beta/{parent}/metadataImports', description=""" Lists imports in a service. """, tags=['metastore_operations_handling', 'metastore_service_configuration'], security=[ UnsuportedSecurityStub(name="None"), UnsuportedSecurityStub(name="None"), ], ) def metastore_projects_locations_services_metadata_imports_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( '/v1beta/{parent}/metadataImports', description=""" Creates a new MetadataImport in a given project and location. """, tags=['metastore_metadata_import_handling'], security=[ UnsuportedSecurityStub(name="None"), UnsuportedSecurityStub(name="None"), ], ) def metastore_projects_locations_services_metadata_imports_create( parent: str, metadata_import_id: Optional[str] = Query(None, alias='metadataImportId'), 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: MetadataImport = None, ): raise RuntimeError("Should be patched by MCPProxy and never executed") @app.get( '/v1beta/{parent}/services', description=""" Lists services in a project and location. """, tags=['metastore_operations_handling', 'metastore_service_configuration'], security=[ UnsuportedSecurityStub(name="None"), UnsuportedSecurityStub(name="None"), ], ) def metastore_projects_locations_services_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( '/v1beta/{parent}/services', description=""" Creates a metastore service in a project and location. """, tags=['metastore_service_configuration', 'metastore_operations_handling'], security=[ UnsuportedSecurityStub(name="None"), UnsuportedSecurityStub(name="None"), ], ) def metastore_projects_locations_services_create( parent: str, request_id: Optional[str] = Query(None, alias='requestId'), service_id: Optional[str] = Query(None, alias='serviceId'), 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: Service = None, ): raise RuntimeError("Should be patched by MCPProxy and never executed") @app.get( '/v1beta/{resource}:getIamPolicy', description=""" Gets the access control policy for a resource. Returns an empty policy if the resource exists and does not have a policy set. """, tags=['database_iam_policy_administration'], security=[ UnsuportedSecurityStub(name="None"), UnsuportedSecurityStub(name="None"), ], ) def get_database_iam_policy( resource: str, options_requested_policy_version: Optional[int] = Query( None, alias='options.requestedPolicyVersion' ), 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( '/v1beta/{resource}:removeIamPolicy', description=""" Removes the attached IAM policies for a resource """, tags=['database_iam_policy_administration'], security=[ UnsuportedSecurityStub(name="None"), UnsuportedSecurityStub(name="None"), ], ) def metastore_projects_locations_services_remove_iam_policy( resource: 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: RemoveIamPolicyRequest = None, ): raise RuntimeError("Should be patched by MCPProxy and never executed") @app.post( '/v1beta/{resource}:setIamPolicy', description=""" Sets the access control policy on the specified resource. Replaces any existing policy.Can return NOT_FOUND, INVALID_ARGUMENT, and PERMISSION_DENIED errors. """, tags=['database_iam_policy_administration'], security=[ UnsuportedSecurityStub(name="None"), UnsuportedSecurityStub(name="None"), ], ) def set_iam_policy_for_database_table( resource: 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: SetIamPolicyRequest = None, ): raise RuntimeError("Should be patched by MCPProxy and never executed") @app.post( '/v1beta/{resource}:testIamPermissions', description=""" Returns permissions that a caller has on the specified resource. If the resource does not exist, this will return an empty set of permissions, not a NOT_FOUND error.Note: This operation is designed to be used for building permission-aware UIs and command-line tools, not for authorization checking. This operation may "fail open" without warning. """, tags=['database_iam_policy_administration'], security=[ UnsuportedSecurityStub(name="None"), UnsuportedSecurityStub(name="None"), ], ) def test_iam_permissions_for_resource( resource: 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: TestIamPermissionsRequest = None, ): raise RuntimeError("Should be patched by MCPProxy and never executed") @app.post( '/v1beta/{service}:alterLocation', description=""" Alter metadata resource location. The metadata resource can be a database, table, or partition. This functionality only updates the parent directory for the respective metadata resource and does not transfer any existing data to the new location. """, tags=[ 'metastore_metadata_import_handling', 'metastore_service_configuration', 'metastore_resource_location_management', ], security=[ UnsuportedSecurityStub(name="None"), UnsuportedSecurityStub(name="None"), ], ) def metastore_projects_locations_services_alter_location( service: 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: AlterMetadataResourceLocationRequest = None, ): raise RuntimeError("Should be patched by MCPProxy and never executed") @app.post( '/v1beta/{service}:exportMetadata', description=""" Exports metadata from a service. """, tags=[ 'metastore_metadata_import_handling', 'metastore_resource_location_management', ], security=[ UnsuportedSecurityStub(name="None"), UnsuportedSecurityStub(name="None"), ], ) def metastore_projects_locations_services_export_metadata( service: 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: ExportMetadataRequest = None, ): raise RuntimeError("Should be patched by MCPProxy and never executed") @app.post( '/v1beta/{service}:moveTableToDatabase', description=""" Move a table to another database. """, tags=['metastore_resource_location_management'], security=[ UnsuportedSecurityStub(name="None"), UnsuportedSecurityStub(name="None"), ], ) def metastore_projects_locations_services_move_table_to_database( service: 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: MoveTableToDatabaseRequest = None, ): raise RuntimeError("Should be patched by MCPProxy and never executed") @app.post( '/v1beta/{service}:queryMetadata', description=""" Query DPMS metadata. """, tags=['metastore_service_configuration', 'database_iam_policy_administration'], security=[ UnsuportedSecurityStub(name="None"), UnsuportedSecurityStub(name="None"), ], ) def metastore_projects_locations_services_query_metadata( service: 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: QueryMetadataRequest = None, ): raise RuntimeError("Should be patched by MCPProxy and never executed") @app.post( '/v1beta/{service}:restore', description=""" Restores a service from a backup. """, tags=['metastore_backup_handling', 'metastore_service_configuration'], security=[ UnsuportedSecurityStub(name="None"), UnsuportedSecurityStub(name="None"), ], ) def metastore_projects_locations_services_restore( service: 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: RestoreServiceRequest = 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)