Access Context Manager API MCP Server

# generated by fastapi-codegen: # filename: openapi.yaml # timestamp: 2025-06-29T00:36:38+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 ( AccessLevel, AccessLevelFormat, AccessPolicy, Alt, FieldXgafv, ListAccessLevelsResponse, ListAccessPoliciesResponse, ListServicePerimetersResponse, Operation, ServicePerimeter, ) app = MCPProxy( contact={'name': 'Google', 'url': 'https://google.com', 'x-twitter': 'youtube'}, description='An API for setting attribute based access control to requests to Google Cloud services.', license={ 'name': 'Creative Commons Attribution 3.0', 'url': 'http://creativecommons.org/licenses/by/3.0/', }, termsOfService='https://developers.google.com/terms/', title='Access Context Manager API', version='v1beta', servers=[{'url': 'https://accesscontextmanager.googleapis.com/'}], ) @app.get( '/v1beta/accessPolicies', description=""" List all AccessPolicies under a container. """, tags=['access_policy_management', 'access_level_management'], security=[ UnsuportedSecurityStub(name="None"), UnsuportedSecurityStub(name="None"), ], ) def accesscontextmanager_access_policies_list( page_size: Optional[int] = Query(None, alias='pageSize'), page_token: Optional[str] = Query(None, alias='pageToken'), parent: Optional[str] = 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.post( '/v1beta/accessPolicies', description=""" Create an `AccessPolicy`. Fails if this organization already has a `AccessPolicy`. The longrunning Operation will have a successful status once the `AccessPolicy` has propagated to long-lasting storage. Syntactic and basic semantic errors will be returned in `metadata` as a BadRequest proto. """, tags=['access_policy_management'], security=[ UnsuportedSecurityStub(name="None"), UnsuportedSecurityStub(name="None"), ], ) def accesscontextmanager_access_policies_create( 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: AccessPolicy = None, ): raise RuntimeError("Should be patched by MCPProxy and never executed") @app.delete( '/v1beta/{name}', description=""" Delete a Service Perimeter by resource name. The longrunning operation from this RPC will have a successful status once the Service Perimeter has been removed from long-lasting storage. """, tags=['access_policy_management', 'access_level_management'], security=[ UnsuportedSecurityStub(name="None"), UnsuportedSecurityStub(name="None"), ], ) def accesscontextmanager_access_policies_service_perimeters_delete( 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.get( '/v1beta/{name}', description=""" Gets the latest state of a long-running operation. Clients can use this method to poll the operation result at intervals as recommended by the API service. """, tags=['access_level_management', 'access_policy_management'], security=[ UnsuportedSecurityStub(name="None"), UnsuportedSecurityStub(name="None"), ], ) def accesscontextmanager_operations_get( name: str, access_level_format: Optional[AccessLevelFormat] = Query( None, alias='accessLevelFormat' ), 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=""" Update a Service Perimeter. The longrunning operation from this RPC will have a successful status once the changes to the Service Perimeter have propagated to long-lasting storage. Service Perimeter containing errors will result in an error response for the first error encountered. """, tags=['access_policy_management'], security=[ UnsuportedSecurityStub(name="None"), UnsuportedSecurityStub(name="None"), ], ) def accesscontextmanager_access_policies_service_perimeters_patch( name: str, 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: ServicePerimeter = None, ): raise RuntimeError("Should be patched by MCPProxy and never executed") @app.get( '/v1beta/{parent}/accessLevels', description=""" List all Access Levels for an access policy. """, tags=['access_level_management', 'access_policy_management'], security=[ UnsuportedSecurityStub(name="None"), UnsuportedSecurityStub(name="None"), ], ) def accesscontextmanager_access_policies_access_levels_list( parent: str, access_level_format: Optional[AccessLevelFormat] = Query( None, alias='accessLevelFormat' ), 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}/accessLevels', description=""" Create an Access Level. The longrunning operation from this RPC will have a successful status once the Access Level has propagated to long-lasting storage. Access Levels containing errors will result in an error response for the first error encountered. """, tags=['access_level_management'], security=[ UnsuportedSecurityStub(name="None"), UnsuportedSecurityStub(name="None"), ], ) def accesscontextmanager_access_policies_access_levels_create( 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: AccessLevel = None, ): raise RuntimeError("Should be patched by MCPProxy and never executed") @app.get( '/v1beta/{parent}/servicePerimeters', description=""" List all Service Perimeters for an access policy. """, tags=['access_policy_management', 'access_level_management'], security=[ UnsuportedSecurityStub(name="None"), UnsuportedSecurityStub(name="None"), ], ) def accesscontextmanager_access_policies_service_perimeters_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( '/v1beta/{parent}/servicePerimeters', description=""" Create a Service Perimeter. The longrunning operation from this RPC will have a successful status once the Service Perimeter has propagated to long-lasting storage. Service Perimeters containing errors will result in an error response for the first error encountered. """, tags=['access_policy_management'], security=[ UnsuportedSecurityStub(name="None"), UnsuportedSecurityStub(name="None"), ], ) def accesscontextmanager_access_policies_service_perimeters_create( 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: ServicePerimeter = 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)