mcp-sqlalchemy

Instructions

Implementation Reference

• mcp_sqlalchemy_server/server.py:287-324 (handler)

  The handler function that executes the SQL query on a pyodbc connection, processes results into a list of dicts with truncated long values, and returns them as a JSONL string.

  def podbc_execute_query(query: str, max_rows: int = 100, params: Optional[Dict[str, Any]] = None,
                    user:Optional[str]=None, password:Optional[str]=None, dsn:Optional[str]=None) -> str:
      """
      Execute a SQL query and return results in JSONL format.
  
      Args:
          query (str): The SQL query to execute.
          max_rows (int): Maximum number of rows to return. Default is 100.
          params (Optional[Dict[str, Any]]): Optional dictionary of parameters to pass to the query.
          user (Optional[str]=None): Optional username.
          password (Optional[str]=None): Optional password.
          dsn (Optional[str]=None): Optional dsn name.
  
      Returns:
          str: Results in JSONL format.
      """
      try:
          with get_connection(True, user, password, dsn) as conn:
              cursor = conn.cursor()
              rs = cursor.execute(query) if params is None else cursor.execute(query, params)
              columns = [column[0] for column in rs.description]            
              results = []
              for row in rs:
                  rs_dict = dict(zip(columns, row))
                  truncated_row = {key: (str(value)[:MAX_LONG_DATA] if value is not None else None) for key, value in rs_dict.items()}
                  results.append(truncated_row)                
                  if len(results) >= max_rows:
                      break
                  
              # Convert the results to JSONL format
              jsonl_results = "\n".join(json.dumps(row) for row in results)
  
              # Return the JSONL formatted results
              return jsonl_results
      except pyodbc.Error as e:
          logging.error(f"Error executing query: {e}")
          raise

• mcp_sqlalchemy_server/server.py:283-286 (registration)

  MCP decorator that registers the podbc_execute_query function as a tool with the specified name and description.

  @mcp.tool(
      name="podbc_execute_query",
      description="Execute a SQL query and return results in JSONL format."
  )

• mcp_sqlalchemy_server/__init__.py:5-11 (registration)

  Imports the tool handler from server.py into the package namespace, exposing it for MCP registration.

  from .server import (
      podbc_get_schemas,
      podbc_get_tables,
      podbc_describe_table,
      podbc_filter_table_names,
      podbc_execute_query,
      podbc_execute_query_md,