mcp-timeplus

Instructions

Implementation Reference

• mcp_timeplus/mcp_server.py:140-164 (handler)

  The main handler for the 'run_sql' MCP tool. It uses a thread pool executor to run queries asynchronously with timeout handling, error management, and structured responses compatible with MCP protocol.

  @mcp.tool()
  def run_sql(query: str):
      """Run a query in a Timeplus database"""
      logger.info(f"Executing query: {query}")
      try:
          future = QUERY_EXECUTOR.submit(execute_query, query)
          try:
              result = future.result(timeout=SELECT_QUERY_TIMEOUT_SECS)
              # Check if we received an error structure from execute_query
              if isinstance(result, dict) and "error" in result:
                  logger.warning(f"Query failed: {result['error']}")
                  # MCP requires structured responses; string error messages can cause
                  # serialization issues leading to BrokenResourceError
                  return {"status": "error", "message": f"Query failed: {result['error']}"}
              return result
          except concurrent.futures.TimeoutError:
              logger.warning(f"Query timed out after {SELECT_QUERY_TIMEOUT_SECS} seconds: {query}")
              future.cancel()
              # Return a properly structured response for timeout errors
              return {"status": "error", "message": f"Query timed out after {SELECT_QUERY_TIMEOUT_SECS} seconds"}
      except Exception as e:
          logger.error(f"Unexpected error in run_select_query: {str(e)}")
          # Catch all other exceptions and return them in a structured format
          # to prevent MCP serialization failures
          return {"status": "error", "message": f"Unexpected error: {str(e)}"}

• mcp_timeplus/mcp_server.py:120-139 (helper)

  Supporting function called by the run_sql handler to execute the SQL query using the Timeplus client, format results as a list of dictionaries, and return structured error responses.

  def execute_query(query: str):
      client = create_timeplus_client()
      try:
          readonly = 1 if config.readonly else 0
          res = client.query(query, settings={"readonly": readonly})
          column_names = res.column_names
          rows = []
          for row in res.result_rows:
              row_dict = {}
              for i, col_name in enumerate(column_names):
                  row_dict[col_name] = row[i]
              rows.append(row_dict)
          logger.info(f"Query returned {len(rows)} rows")
          return rows
      except Exception as err:
          logger.error(f"Error executing query: {err}")
          # Return a structured dictionary rather than a string to ensure proper serialization
          # by the MCP protocol. String responses for errors can cause BrokenResourceError.
          return {"error": str(err)}