BC Data Catalogue API MCP Server

# generated by fastapi-codegen: # filename: openapi.yaml # timestamp: 2025-06-29T03:12:41+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 ( APIKeyHeader, BaseSecurity, UnsuportedSecurityStub, ) app = MCPProxy( contact={ 'email': 'data@gov.bc.ca', 'name': 'Data BC', 'url': 'http://data.gov.bc.ca/', }, description='This API provides live access to the BC Data Catalogue. Further documentation on the API is available from http://docs.ckan.org/en/latest/ Confirm the version of the API available from the catalogue by requesting https://catalogue.data.gov.bc.ca/api/3/action/status_show. \n\nPlease note that you may experience issues when submitting requests to the delivery or test environment if using this [OpenAPI specification](https://github.com/bcgov/api-specs) in other API console viewers.', license={ 'name': 'Open Government License - British Columbia', 'url': 'https://www2.gov.bc.ca/gov/content?id=A519A56BC2BF44E4A008B33FCF527F61', }, termsOfService='https://www2.gov.bc.ca/gov/content?id=D1EE0A405E584363B205CD4353E02C88', title='BC Data Catalogue API', version='3.0.1', servers=[ {'description': 'Production', 'url': 'https://catalogue.data.gov.bc.ca/api/3'}, {'description': 'Test', 'url': 'https://cat.data.gov.bc.ca/api/3'}, {'description': 'Delivery', 'url': 'https://cad.data.gov.bc.ca/api/3'}, ], ) @app.get( '/action/organization_activity_list', description=""" Return an organization's activity stream """, tags=['organization_activity_management', 'organization_management'], security=[ UnsuportedSecurityStub(name="None"), APIKeyHeader(name="ckan_api_key"), ], ) def get_action_organization_activity_list( id: Optional[str] = 'ministry-of-agriculture', ): """ Get the activity stream of an organization """ raise RuntimeError("Should be patched by MCPProxy and never executed") @app.get( '/action/organization_activity_list_html', description=""" Return an organization's activity stream as HTML """, tags=['organization_activity_management', 'organization_management'], security=[ UnsuportedSecurityStub(name="None"), APIKeyHeader(name="ckan_api_key"), ], ) def get_action_organization_activity_list_html( id: Optional[str] = 'ministry-of-agriculture', ): """ Get the activity stream of an organization, HTML format """ raise RuntimeError("Should be patched by MCPProxy and never executed") @app.get( '/action/organization_autocomplete', description=""" Return a list of organization names that contain a string """, tags=['organization_management'], security=[ UnsuportedSecurityStub(name="None"), APIKeyHeader(name="ckan_api_key"), ], ) def get_action_organization_autocomplete( q: Optional[str] = 'ministry', limit: Optional[int] = 20 ): """ Get names of organizations that match a query string """ raise RuntimeError("Should be patched by MCPProxy and never executed") @app.get( '/action/organization_follower_count', description=""" Return the number of followers of an organization """, tags=['organization_activity_management', 'organization_management'], security=[ UnsuportedSecurityStub(name="None"), APIKeyHeader(name="ckan_api_key"), ], ) def get_action_organization_follower_count( id: Optional[str] = 'ministry-of-agriculture', ): """ Get number of followers of an organization """ raise RuntimeError("Should be patched by MCPProxy and never executed") @app.get( '/action/organization_follower_list', description=""" Return a list of users that are following a given organization """, tags=['organization_activity_management', 'organization_management'], security=[ UnsuportedSecurityStub(name="None"), APIKeyHeader(name="ckan_api_key"), ], ) def get_action_organization_follower_list( id: Optional[str] = 'ministry-of-agriculture', ): """ Get users following an organization """ raise RuntimeError("Should be patched by MCPProxy and never executed") @app.get( '/action/organization_list', description=""" Returns the names of all indexed organizations """, tags=['organization_management'], security=[ UnsuportedSecurityStub(name="None"), APIKeyHeader(name="ckan_api_key"), ], ) def get_action_organization_list(offset: Optional[int] = 0, limit: Optional[int] = 100): """ Get names of all organizations """ raise RuntimeError("Should be patched by MCPProxy and never executed") @app.get( '/action/organization_list_for_user', description=""" Return the organizations that the user has a given permission for """, tags=['organization_management'], security=[ UnsuportedSecurityStub(name="None"), APIKeyHeader(name="ckan_api_key"), ], ) def get_action_organization_list_for_user(permission: Optional[str] = '"edit_group"'): """ Get organizations that a user has a given permission for """ raise RuntimeError("Should be patched by MCPProxy and never executed") @app.get( '/action/organization_revision_list', description=""" Return an organization's revisions """, tags=['organization_activity_management', 'organization_management'], security=[ UnsuportedSecurityStub(name="None"), APIKeyHeader(name="ckan_api_key"), ], ) def get_action_organization_revision_list( id: Optional[str] = 'ministry-of-agriculture', ): """ Get organization revisions """ raise RuntimeError("Should be patched by MCPProxy and never executed") @app.get( '/action/organization_show', description=""" Return the details of an organization """, tags=['organization_management', 'organization_activity_management'], security=[ UnsuportedSecurityStub(name="None"), APIKeyHeader(name="ckan_api_key"), ], ) def get_action_organization_show( id: Optional[str] = 'ministry-of-agriculture', include_datasets: Optional[bool] = True, ): """ Get details of a specific organization """ raise RuntimeError("Should be patched by MCPProxy and never executed") @app.get( '/action/package_activity_list', description=""" Returns a package's activity stream """, tags=['package_activity_management', 'package_management'], security=[ UnsuportedSecurityStub(name="None"), APIKeyHeader(name="ckan_api_key"), ], ) def get_action_package_activity_list( id: Optional[str] = 'grizzly-bear-population-units', offset: Optional[int] = 0, limit: Optional[int] = 31, ): """ Get the activity stream of a package (dataset) """ raise RuntimeError("Should be patched by MCPProxy and never executed") @app.get( '/action/package_activity_list_html', description=""" The activity stream is rendered as a snippet of HTML meant to be included in an HTML pag, i.e it doesn't have any header or footer. """, tags=['package_activity_management', 'package_management'], security=[ UnsuportedSecurityStub(name="None"), APIKeyHeader(name="ckan_api_key"), ], ) def get_action_package_activity_list_html( id: Optional[str] = 'grizzly-bear-population-units', offset: Optional[int] = 0, limit: Optional[int] = 31, ): """ Get the activity stream of a package (dataset), HTML format """ raise RuntimeError("Should be patched by MCPProxy and never executed") @app.get( '/action/package_autocomplete', description=""" Return a list of datasets that match a string """, tags=['package_management', 'resource_metadata_management'], security=[ UnsuportedSecurityStub(name="None"), APIKeyHeader(name="ckan_api_key"), ], ) def get_action_package_autocomplete( q: Optional[str] = '"Okanagan Lake"', limit: Optional[int] = 10 ): """ Find packages (datasets) matching a query """ raise RuntimeError("Should be patched by MCPProxy and never executed") @app.get( '/action/package_list', description=""" Returns the names of all indexed packages (datasets) """, tags=['package_management', 'package_activity_management'], security=[ UnsuportedSecurityStub(name="None"), APIKeyHeader(name="ckan_api_key"), ], ) def get_action_package_list(offset: Optional[int] = 0, limit: Optional[int] = 100): """ Get a list of all packages (datasets) """ raise RuntimeError("Should be patched by MCPProxy and never executed") @app.get( '/action/package_relationships_list', description=""" Return a dataset's relationships """, tags=['package_activity_management', 'package_management'], security=[ UnsuportedSecurityStub(name="None"), APIKeyHeader(name="ckan_api_key"), ], ) def get_action_package_relationships_list( id: Optional[str] = 'grizzly-bear-population-units', id2: Optional[str] = 'important-fossil-areas', rel: Optional[str] = None, ): """ Get package (dataset) relationships """ raise RuntimeError("Should be patched by MCPProxy and never executed") @app.get( '/action/package_revision_list', description=""" Return a dataset's revision as a list of dictionaries """, tags=['package_activity_management', 'package_management'], security=[ UnsuportedSecurityStub(name="None"), APIKeyHeader(name="ckan_api_key"), ], ) def get_action_package_revision_list( id: Optional[str] = 'grizzly-bear-population-units', ): """ Get list of revisions for a package (dataset) """ raise RuntimeError("Should be patched by MCPProxy and never executed") @app.get( '/action/package_search', description=""" Searches for packages (datasets) matching the specified query terms """, tags=['package_management', 'resource_metadata_management'], security=[ UnsuportedSecurityStub(name="None"), APIKeyHeader(name="ckan_api_key"), ], ) def get_action_package_search(q: Optional[str] = '"Okanagan Lake"'): """ Find packages (datasets) matching query terms """ raise RuntimeError("Should be patched by MCPProxy and never executed") @app.get( '/action/package_show', description=""" Returns metadata about the package (dataset) corresponding to the specified unique name """, tags=['package_management', 'resource_metadata_management'], security=[ UnsuportedSecurityStub(name="None"), APIKeyHeader(name="ckan_api_key"), ], ) def get_action_package_show(id: Optional[str] = 'grizzly-bear-population-units'): """ Get metadata about one specific package (dataset) """ raise RuntimeError("Should be patched by MCPProxy and never executed") @app.get( '/action/related_list', description=""" Returns a dataset's related items. """, tags=['package_activity_management', 'package_management'], security=[ UnsuportedSecurityStub(name="None"), APIKeyHeader(name="ckan_api_key"), ], ) def get_action_related_list( id: Optional[str] = None, dataset: Optional[str] = None, type_filter: Optional[str] = None, sort: Optional[str] = None, featured: Optional[str] = None, ): """ Gets items related to a package (dataset) """ raise RuntimeError("Should be patched by MCPProxy and never executed") @app.get( '/action/resource_search', description=""" Returns a dictionary with two fields ``count`` and ``results``. The ``count`` field contains the total number of Resources found without the limit or query parameters having an effect. The ``results`` field is a list of dictized Resource objects. The query parameter is a required field. It is a string in the form ``{field}:{term}`` or a list of strings, each of the same form. Within each string, ``{field}`` is a field or extra field on the Resource domain object. """, tags=['resource_metadata_management', 'package_management'], security=[ UnsuportedSecurityStub(name="None"), APIKeyHeader(name="ckan_api_key"), ], ) def get_action_resource_search( query: Optional[str] = 'format:csv', fields: Optional[str] = None, order_by: Optional[str] = None, offset: Optional[int] = 0, limit: Optional[int] = 0, ): """ Find resources """ raise RuntimeError("Should be patched by MCPProxy and never executed") @app.get( '/action/resource_show', description=""" Return the metadata of a resource """, tags=['resource_metadata_management', 'package_management'], security=[ UnsuportedSecurityStub(name="None"), APIKeyHeader(name="ckan_api_key"), ], ) def get_action_resource_show( id: Optional[str] = 'e6c8bb1d-3726-418b-9b7e-1d97a9bbb817', include_tracking: Optional[bool] = False, ): """ Get metadata for a specific resource """ raise RuntimeError("Should be patched by MCPProxy and never executed") @app.get( '/action/status_show', description=""" Returns the site status """, tags=['site_health_status'], security=[ UnsuportedSecurityStub(name="None"), APIKeyHeader(name="ckan_api_key"), ], ) def get_action_status_show(): """ Get the site status """ raise RuntimeError("Should be patched by MCPProxy and never executed") @app.get( '/action/tag_list', description=""" Returns the names of all indexed tags """, tags=['tagging_system_management'], security=[ UnsuportedSecurityStub(name="None"), APIKeyHeader(name="ckan_api_key"), ], ) def get_action_tag_list(offset: Optional[int] = 0, limit: Optional[int] = 100): """ Get a list of tags """ 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)