mcp-sqlalchemy

Instructions

Implementation Reference

• mcp_sqlalchemy_server/server.py:414-443 (handler)

  The main handler function for the 'podbc_spasql_query' tool, including the @mcp.tool decorator which also serves as registration. It connects to the database using pyodbc, executes a SPASQL query via a specific Virtuoso command, and returns the result.

  @mcp.tool(
      name="podbc_spasql_query",
      description="Execute a SPASQL query and return results."
  )
  def podbc_spasql_query(query: str, max_rows:Optional[int] = 20, timeout:Optional[int] = 300000,
                      user:Optional[str]=None, password:Optional[str]=None, dsn:Optional[str]=None) -> str:
      """
      Execute a SPASQL query and return results in JSONL format.
  
      Args:
          query (str): The SPASQL query to execute.
          max_rows (int): Maximum number of rows to return. Default is 100.
          timeout (int): Query timeout. Default is 30000ms.
          user (Optional[str]=None): Optional username.
          password (Optional[str]=None): Optional password.
          dsn (Optional[str]=None): Optional dsn name.
  
      Returns:
          str: Results in requested format as string.
      """
      try:
          with get_connection(True, user, password, dsn) as conn:
              cursor = conn.cursor()
              cmd = f"select Demo.demo.execute_spasql_query(charset_recode(?, '_WIDE_', 'UTF-8'), ?, ?) as result"
              rs = cursor.execute(cmd, (query, max_rows, timeout,)).fetchone()
              return rs[0]
      except pyodbc.Error as e:
          logging.error(f"Error executing query: {e}")
          raise

• mcp_sqlalchemy_server/__init__.py:33-33 (registration)

  The tool function is exported in the package's __all__ list for easy import.

  "podbc_spasql_query",

• mcp_sqlalchemy_server/__init__.py:14-14 (registration)

  The tool function is imported from server.py into the package namespace.

  podbc_spasql_query,