run_sql

Instructions

Implementation Reference

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

  The run_sql tool handler - decorated with @mcp.tool(), executes a SQL query against Timeplus database with timeout handling and error management.

  @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-138 (helper)

  execute_query helper function - creates a Timeplus client, runs the query, and returns structured results or error dictionaries.

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

• mcp_timeplus/mcp_server.py:141-141 (schema)

  The run_sql tool accepts a single 'query' parameter of type str, no explicit schema/Pydantic model.

  def run_sql(query: str):

• mcp_timeplus/mcp_server.py:140-140 (registration)

  Registration via @mcp.tool() decorator on the run_sql function.

  @mcp.tool()

• mcp_timeplus/__init__.py:1-5 (registration)

  run_sql is exported from the package's __init__.py via import from mcp_server.

  from .mcp_server import (
      create_timeplus_client,
      list_databases,
      list_tables,
      run_sql,