mcp-sqlalchemy

Instructions

Implementation Reference

• mcp_sqlalchemy_server/server.py:53-56 (registration)

  The @mcp.tool decorator registers the 'podbc_get_schemas' tool with its name and description.

  @mcp.tool(
      name="podbc_get_schemas",
      description="Retrieve and return a list of all schema names from the connected database."
  )

• mcp_sqlalchemy_server/server.py:57-79 (handler)

  The handler function that executes the logic: connects to the ODBC database, queries for tables to get unique catalogs (schemas), and returns them as JSON string.

  def podbc_get_schemas(user:Optional[str]=None, password:Optional[str]=None, dsn:Optional[str]=None) -> str:
      """
      Retrieve and return a list of all schema names from the connected database.
  
      Args:
          user (Optional[str]=None): Optional username.
          password (Optional[str]=None): Optional password.
          dsn (Optional[str]=None): Optional dsn name.
  
      Returns:
          str: A list of schema names.
      """
      try:
          with get_connection(True, user, password, dsn) as conn:
              cursor = conn.cursor()
              rs = cursor.tables(table=None, catalog="%", schema=None, tableType=None);
              catalogs = {row[0] for row in rs.fetchall()}
              return json.dumps(list(catalogs))
  
      except pyodbc.Error as e:
          logging.error(f"Error retrieving schemas: {e}")
          raise

• mcp_sqlalchemy_server/server.py:57-68 (schema)

  Input schema defined by function parameters with type hints and documentation; output is str (JSON list of schemas).

  def podbc_get_schemas(user:Optional[str]=None, password:Optional[str]=None, dsn:Optional[str]=None) -> str:
      """
      Retrieve and return a list of all schema names from the connected database.
  
      Args:
          user (Optional[str]=None): Optional username.
          password (Optional[str]=None): Optional password.
          dsn (Optional[str]=None): Optional dsn name.
  
      Returns:
          str: A list of schema names.
      """

• mcp_sqlalchemy_server/server.py:27-44 (helper)

  Helper function to establish pyodbc connection using env vars or provided credentials, used by the tool.

  def get_connection(readonly=True, uid: Optional[str] = None, pwd: Optional[str] = None, 
                  dsn: Optional[str] = None) -> pyodbc.Connection:
      dsn = DB_DSN if dsn is None else dsn
      uid = DB_UID if uid is None else uid
      pwd = DB_PWD if pwd is None else pwd
  
      if dsn is None:
          raise ValueError("ODBC_DSN environment variable is not set.")
      if uid is None:
          raise ValueError("ODBC_USER environment variable is not set.")
      if pwd is None:
          raise ValueError("ODBC_PASSWORD environment variable is not set.")
  
      dsn_string = f"DSN={dsn};UID={uid};PWD={pwd}"
      logging.info(f"DSN:{dsn}  UID:{uid}")
      # connection_string="DSN=VOS;UID=dba;PWD=dba"
  
      return pyodbc.connect(dsn_string, autocommit=True, readonly=readonly)