mcp-timeplus

Instructions

Implementation Reference

• mcp_timeplus/mcp_server.py:52-119 (handler)

  The main handler function for the 'list_tables' MCP tool. It is decorated with @mcp.tool() which registers it as an MCP tool. The function lists tables/streams in a specified Timeplus database, optionally filtered by 'like' pattern. It fetches table lists, comments, column comments, schema details, and CREATE statements for each table, compiling a detailed list of table metadata.

  @mcp.tool()
  def list_tables(database: str = 'default', like: str = None):
      """List available tables/streams in the given database"""
      logger.info(f"Listing tables in database '{database}'")
      client = create_timeplus_client()
      query = f"SHOW STREAMS FROM {quote_identifier(database)}"
      if like:
          query += f" LIKE {format_query_value(like)}"
      result = client.command(query)
  
      # Get all table comments in one query
      table_comments_query = f"SELECT name, comment FROM system.tables WHERE database = {format_query_value(database)}"
      table_comments_result = client.query(table_comments_query)
      table_comments = {row[0]: row[1] for row in table_comments_result.result_rows}
  
      # Get all column comments in one query
      column_comments_query = f"SELECT table, name, comment FROM system.columns WHERE database = {format_query_value(database)}"
      column_comments_result = client.query(column_comments_query)
      column_comments = {}
      for row in column_comments_result.result_rows:
          table, col_name, comment = row
          if table not in column_comments:
              column_comments[table] = {}
          column_comments[table][col_name] = comment
  
      def get_table_info(table):
          logger.info(f"Getting schema info for table {database}.{table}")
          schema_query = f"DESCRIBE STREAM {quote_identifier(database)}.{quote_identifier(table)}"
          schema_result = client.query(schema_query)
  
          columns = []
          column_names = schema_result.column_names
          for row in schema_result.result_rows:
              column_dict = {}
              for i, col_name in enumerate(column_names):
                  column_dict[col_name] = row[i]
              # Add comment from our pre-fetched comments
              if table in column_comments and column_dict['name'] in column_comments[table]:
                  column_dict['comment'] = column_comments[table][column_dict['name']]
              else:
                  column_dict['comment'] = None
              columns.append(column_dict)
  
          create_table_query = f"SHOW CREATE STREAM {database}.`{table}`"
          create_table_result = client.command(create_table_query)
  
          return {
              "database": database,
              "name": table,
              "comment": table_comments.get(table),
              # "columns": columns, # exclude columns in the output since it's too verbose, the DDL below has enough information
              "create_table_query": create_table_result,
          }
  
      tables = []
      if isinstance(result, str):
          # Single table result
          for table in (t.strip() for t in result.split()):
              if table:
                  tables.append(get_table_info(table))
      elif isinstance(result, Sequence):
          # Multiple table results
          for table in result:
              tables.append(get_table_info(table))
  
      logger.info(f"Found {len(tables)} tables")
      return tables