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)}