mcp-sqlalchemy

Instructions

Implementation Reference

• mcp_sqlalchemy_server/server.py:326-373 (handler)

  The @mcp.tool decorator registers the tool, and the function podbc_execute_query_md implements the logic: connects to DB via pyodbc, executes the query, truncates long values, formats results as Markdown table, limits to max_rows.

  @mcp.tool(
      name="podbc_execute_query_md",
      description="Execute a SQL query and return results in Markdown table format."
  )
  def podbc_execute_query_md(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 Markdown table 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 Markdown table 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
                  
              # Create the Markdown table header
              md_table = "| " + " | ".join(columns) + " |\n"
              md_table += "| " + " | ".join(["---"] * len(columns)) + " |\n"
  
              # Add rows to the Markdown table
              for row in results:
                  md_table += "| " + " | ".join(str(row[col]) for col in columns) + " |\n"
  
              # Return the Markdown formatted results
              return md_table
  
      except pyodbc.Error as e:
          logging.error(f"Error executing query: {e}")
          raise

• mcp_sqlalchemy_server/server.py:327-327 (registration)

  The tool is registered with FastMCP using the @mcp.tool decorator specifying the name.

  name="podbc_execute_query_md",

• mcp_sqlalchemy_server/server.py:330-331 (schema)

  Input schema defined by function parameters with type hints: query (str, required), max_rows (int=100), params (Optional[Dict]), user/password/dsn optional.

  def podbc_execute_query_md(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: