Keboola Explorer MCP Server

from typing import Literal, cast from pydantic import BaseModel, ConfigDict, Field from keboola_mcp_server.clients.client import ( CONDITIONAL_FLOW_COMPONENT_ID, DATA_APP_COMPONENT_ID, FLOW_TYPES, FlowType, KeboolaClient, ) URLType = Literal['ui-detail', 'ui-dashboard', 'docs'] class Link(BaseModel): model_config = ConfigDict(frozen=True) type: URLType = Field(..., description='The type of the URL.') title: str = Field(..., description='The name of the URL.') url: str = Field(..., description='The URL.') @classmethod def detail(cls, title: str, url: str) -> 'Link': return cls(type='ui-detail', title=title, url=url) @classmethod def dashboard(cls, title: str, url: str) -> 'Link': return cls(type='ui-dashboard', title=title, url=url) @classmethod def docs(cls, title: str, url: str) -> 'Link': return cls(type='docs', title=title, url=url) class ProjectLinksManager: FLOW_DOCUMENTATION_URL = 'https://help.keboola.com/flows/' def __init__(self, *, base_url: str, project_id: str, branch_id: str | None): self._base_url = base_url self._project_id = project_id self._branch_id = branch_id @classmethod async def from_client(cls, client: KeboolaClient) -> 'ProjectLinksManager': project_id = await client.storage_client.project_id() return cls(base_url=client.storage_api_url, project_id=project_id, branch_id=client.branch_id) def _url(self, path: str) -> str: parts = [self._base_url, 'admin/projects', self._project_id] if self._branch_id: parts += ['branch', self._branch_id] parts.append(path) return '/'.join(parts) @staticmethod def _flow_type_from_component_id(component_id: str | None) -> FlowType | None: if component_id in FLOW_TYPES: return cast(FlowType, component_id) return None @staticmethod def _is_data_app_component(component_id: str | None) -> bool: return component_id == DATA_APP_COMPONENT_ID @staticmethod def _is_transformation_component(component_id: str) -> bool: return bool(component_id and 'transformation' in component_id) def get_links( self, *, bucket_id: str | None = None, table_id: str | None = None, component_id: str | None = None, configuration_id: str | None = None, name: str | None = None, ) -> list[Link]: """ Get the most relevant links for Keboola Objects based on the provided mutually exclusive identifiers. """ if component_id and configuration_id: return [ self.get_component_config_link( component_id=component_id, configuration_id=configuration_id, configuration_name=name or '', ) ] if component_id: return [self.get_config_dashboard_link(component_id=component_id, component_name=name or '')] if table_id: return [self.get_table_detail_link_from_table_id(table_id=table_id)] if bucket_id: return [self.get_bucket_detail_link(bucket_id=bucket_id, bucket_name=name or bucket_id)] return [] # --- Project --- def get_project_detail_link(self) -> Link: return Link.detail(title='Project Dashboard', url=self._url('')) def get_project_links(self) -> list[Link]: return [self.get_project_detail_link()] # --- Flows --- def get_flow_detail_link(self, flow_id: str | int, flow_name: str, flow_type: FlowType) -> Link: """Get detail link for a specific flow based on its type.""" flow_path = 'flows-v2' if flow_type == CONDITIONAL_FLOW_COMPONENT_ID else 'flows' return Link.detail(title=f'Flow: {flow_name}', url=self._url(f'{flow_path}/{flow_id}')) def get_flows_dashboard_link(self, flow_type: FlowType) -> Link: """Get dashboard link for flows based on the flow type.""" flow_path = 'flows-v2' if flow_type == CONDITIONAL_FLOW_COMPONENT_ID else 'flows' flow_label = 'Conditional Flows' if flow_type == CONDITIONAL_FLOW_COMPONENT_ID else 'Flows' return Link.dashboard(title=f'{flow_label} in the project', url=self._url(flow_path)) def get_flows_docs_link(self) -> Link: return Link.docs(title='Documentation for Keboola Flows', url=self.FLOW_DOCUMENTATION_URL) def get_flow_links(self, flow_id: str | int, flow_name: str, flow_type: FlowType) -> list[Link]: """Get all relevant links for a flow based on its type.""" return [ self.get_flow_detail_link(flow_id, flow_name, flow_type), self.get_flows_dashboard_link(flow_type), self.get_flows_docs_link(), ] # --- Components --- def get_component_config_link( self, component_id: str, configuration_id: str, configuration_name: str, ) -> Link: """ Get the link to the configuration of a component based on its type, transformation, data app, flow, or component. """ if self._is_transformation_component(component_id): return self.get_transformation_config_link( transformation_type=component_id, transformation_id=configuration_id, transformation_name=configuration_name, ) if self._is_data_app_component(component_id): return self.get_data_app_config_link( configuration_id=configuration_id, configuration_name=configuration_name, uses_basic_authentication=False, ) if flow_type := self._flow_type_from_component_id(component_id): return self.get_flow_detail_link( flow_id=configuration_id, flow_name=configuration_name, flow_type=flow_type ) return Link.detail( title=f'Configuration: {configuration_name}', url=self._url(f'components/{component_id}/{configuration_id}') ) def get_config_dashboard_link(self, component_id: str, component_name: str | None) -> Link: component_name = f'{component_name}' if component_name else f'Component "{component_id}"' return Link.dashboard( title=f'{component_name} Configurations Dashboard', url=self._url(f'components/{component_id}') ) def get_used_components_link(self) -> Link: return Link.dashboard(title='Used Components Dashboard', url=self._url('components/configurations')) def get_configuration_links(self, component_id: str, configuration_id: str, configuration_name: str) -> list[Link]: return [ self.get_component_config_link( component_id=component_id, configuration_id=configuration_id, configuration_name=configuration_name ), self.get_config_dashboard_link(component_id=component_id, component_name=None), ] # --- Data Apps --- def get_data_app_config_link( self, configuration_id: str, configuration_name: str, uses_basic_authentication: bool ) -> Link: title = ( f'Data App Configuration (To see password, click on "OPEN DATA APP"): {configuration_name}' if uses_basic_authentication else f'Data App Configuration: {configuration_name}' ) return Link.detail(title=title, url=self._url(f'data-apps/{configuration_id}')) def get_data_app_dashboard_link(self) -> Link: return Link.dashboard(title='Data Apps in the project', url=self._url('data-apps')) def get_data_app_deployment_link(self, deployment_link: str) -> Link: return Link.detail(title='Data App Deployment', url=deployment_link) def get_data_app_links( self, configuration_id: str, configuration_name: str, deployment_link: str | None = None, uses_basic_authentication: bool = False, ) -> list[Link]: links = [ self.get_data_app_config_link( configuration_id=configuration_id, configuration_name=configuration_name, uses_basic_authentication=uses_basic_authentication, ), self.get_data_app_dashboard_link(), ] if deployment_link: links.append(self.get_data_app_deployment_link(deployment_link)) return links # --- Transformations --- def get_transformations_dashboard_link(self) -> Link: return Link.dashboard(title='Transformations dashboard', url=self._url('transformations-v2')) def get_transformation_config_link( self, transformation_type: str, transformation_id: str, transformation_name: str ) -> Link: return Link.detail( title=f'Transformation: {transformation_name}', url=self._url(f'transformations-v2/{transformation_type}/{transformation_id}'), ) def get_transformation_links( self, transformation_type: str, transformation_id: str, transformation_name: str ) -> list[Link]: return [ self.get_transformation_config_link(transformation_type, transformation_id, transformation_name), self.get_transformations_dashboard_link(), ] # --- Jobs --- def get_job_detail_link(self, job_id: str) -> Link: return Link.detail(title=f'Job: {job_id}', url=self._url(f'queue/{job_id}')) def get_jobs_dashboard_link(self) -> Link: return Link.dashboard(title='Jobs in the project', url=self._url('queue')) def get_job_links(self, job_id: str) -> list[Link]: return [self.get_job_detail_link(job_id), self.get_jobs_dashboard_link()] # --- Buckets --- def get_bucket_detail_link(self, bucket_id: str, bucket_name: str) -> Link: return Link.detail(title=f'Bucket: {bucket_name}', url=self._url(f'storage/{bucket_id}')) def get_bucket_dashboard_link(self) -> Link: return Link.dashboard(title='Buckets in the project', url=self._url('storage')) def get_bucket_links(self, bucket_id: str, bucket_name: str) -> list[Link]: return [ self.get_bucket_detail_link(bucket_id, bucket_name), self.get_bucket_dashboard_link(), ] # --- Tables --- def get_table_detail_link(self, bucket_id: str, table_name: str) -> Link: return Link.detail(title=f'Table: {table_name}', url=self._url(f'storage/{bucket_id}/table/{table_name}')) def get_table_detail_link_from_table_id(self, table_id: str) -> Link: table_name = table_id.split('.')[-1] bucket_id = '.'.join(table_id.split('.')[:-1]) return self.get_table_detail_link(bucket_id=bucket_id, table_name=table_name) def get_table_links(self, bucket_id: str, bucket_name: str, table_name: str) -> list[Link]: return [ self.get_table_detail_link(bucket_id, table_name), self.get_bucket_detail_link(bucket_id=bucket_id, bucket_name=bucket_name), ]