Digital Asset Links API MCP Server

# generated by fastapi-codegen: # filename: openapi.yaml # timestamp: 2025-06-29T01:54:01+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, BulkCheckRequest, BulkCheckResponse, CheckResponse, FieldXgafv, ListResponse, ) app = MCPProxy( contact={'name': 'Google', 'url': 'https://google.com', 'x-twitter': 'youtube'}, description='Discovers relationships between online assets such as websites or mobile apps.', license={ 'name': 'Creative Commons Attribution 3.0', 'url': 'http://creativecommons.org/licenses/by/3.0/', }, termsOfService='https://developers.google.com/terms/', title='Digital Asset Links API', version='v1', servers=[{'url': 'https://digitalassetlinks.googleapis.com/'}], ) @app.post( '/v1/assetlinks:bulkCheck', description=""" Send a bundle of statement checks in a single RPC to minimize latency and service load. Statements need not be all for the same source and/or target. We recommend using this method when you need to check more than one statement in a short period of time. """, tags=['digital_asset_link_management'], ) def digitalassetlinks_assetlinks_bulk_check( 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: BulkCheckRequest = None, ): raise RuntimeError("Should be patched by MCPProxy and never executed") @app.get( '/v1/assetlinks:check', description=""" Determines whether the specified (directional) relationship exists between the specified source and target assets. The relation describes the intent of the link between the two assets as claimed by the source asset. An example for such relationships is the delegation of privileges or permissions. This command is most often used by infrastructure systems to check preconditions for an action. For example, a client may want to know if it is OK to send a web URL to a particular mobile app instead. The client can check for the relevant asset link from the website to the mobile app to decide if the operation should be allowed. A note about security: if you specify a secure asset as the source, such as an HTTPS website or an Android app, the API will ensure that any statements used to generate the response have been made in a secure way by the owner of that asset. Conversely, if the source asset is an insecure HTTP website (that is, the URL starts with `http://` instead of `https://`), the API cannot verify its statements securely, and it is not possible to ensure that the website's statements have not been altered by a third party. For more information, see the [Digital Asset Links technical design specification](https://github.com/google/digitalassetlinks/blob/master/well-known/details.md). """, tags=['digital_asset_link_management'], ) def digitalassetlinks_assetlinks_check( relation: Optional[str] = None, source_android_app_certificate_sha256_fingerprint: Optional[str] = Query( None, alias='source.androidApp.certificate.sha256Fingerprint' ), source_android_app_package_name: Optional[str] = Query( None, alias='source.androidApp.packageName' ), source_web_site: Optional[str] = Query(None, alias='source.web.site'), target_android_app_certificate_sha256_fingerprint: Optional[str] = Query( None, alias='target.androidApp.certificate.sha256Fingerprint' ), target_android_app_package_name: Optional[str] = Query( None, alias='target.androidApp.packageName' ), target_web_site: Optional[str] = Query(None, alias='target.web.site'), 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/statements:list', description=""" Retrieves a list of all statements from a given source that match the specified target and statement string. The API guarantees that all statements with secure source assets, such as HTTPS websites or Android apps, have been made in a secure way by the owner of those assets, as described in the [Digital Asset Links technical design specification](https://github.com/google/digitalassetlinks/blob/master/well-known/details.md). Specifically, you should consider that for insecure websites (that is, where the URL starts with `http://` instead of `https://`), this guarantee cannot be made. The `List` command is most useful in cases where the API client wants to know all the ways in which two assets are related, or enumerate all the relationships from a particular source asset. Example: a feature that helps users navigate to related items. When a mobile app is running on a device, the feature would make it easy to navigate to the corresponding web site or Google+ profile. """, tags=['digital_asset_link_management'], ) def digitalassetlinks_statements_list( relation: Optional[str] = None, source_android_app_certificate_sha256_fingerprint: Optional[str] = Query( None, alias='source.androidApp.certificate.sha256Fingerprint' ), source_android_app_package_name: Optional[str] = Query( None, alias='source.androidApp.packageName' ), source_web_site: Optional[str] = Query(None, alias='source.web.site'), 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)