podbc_describe_table
Retrieve table definitions, including column names, data types, and keys, from any SQLAlchemy-accessible DBMS using the MCP server's pyodbc connectivity.
Instructions
Retrieve and return a dictionary containing the definition of a table, including column names, data types, nullable, autoincrement, primary key, and foreign keys.
Input Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| Schema | Yes | ||
| table | Yes | ||
| url | No |
Implementation Reference
- mcp_sqlalchemy_server/server.py:117-153 (handler)Main handler function for podbc_describe_table tool. Handles connection, table existence check, and delegates to _get_table_info for details.
@mcp.tool( name="podbc_describe_table", description="Retrieve and return a dictionary containing the definition of a table, including column names, data types, nullable," " autoincrement, primary key, and foreign keys." ) def podbc_describe_table(Schema:str, table: str, user:Optional[str]=None, password:Optional[str]=None, dsn:Optional[str]=None) -> str: """ Retrieve and return a dictionary containing the definition of a table, including column names, data types, nullable, autoincrement, primary key, and foreign keys. If `schema` is None, returns the table definition for the specified table in all schemas. If `schema` is not None, returns the table definition for the specified table in the specified schema. Args: schema (str): The name of the schema to retrieve the table definition for. If None, retrieves the table definition for all schemas. table (str): The name of the table to retrieve the definition for. user (Optional[str]=None): Optional username. password (Optional[str]=None): Optional password. dsn (Optional[str]=None): Optional dsn name. Returns: str: A dictionary containing the table definition, including column names, data types, nullable, autoincrement, primary key, and foreign keys. """ cat = "%" if Schema is None else Schema table_definition = {} try: with get_connection(True, user, password, dsn) as conn: rc, tbl = _has_table(conn, cat=cat, table=table) if rc: table_definition = _get_table_info(conn, cat=tbl.get("cat"), sch=tbl.get("sch"), table=tbl.get("name")) return json.dumps(table_definition, indent=2) except pyodbc.Error as e: logging.error(f"Error retrieving table definition: {e}") raise - Core helper that fetches columns, primary keys, foreign keys, assembles table definition, marks PK columns.
def _get_table_info(conn, cat:str, sch: str, table: str) -> Dict[str, Any]: try: columns = _get_columns(conn, cat=cat, sch=sch, table=table) primary_keys = _get_pk_constraint(conn, cat=cat, sch=sch, table=table)['constrained_columns'] foreign_keys = _get_foreign_keys(conn, cat=cat, sch=sch, table=table) table_info = { "TABLE_CAT": cat, "TABLE_SCHEM": sch, "TABLE_NAME": table, "columns": columns, "primary_keys": primary_keys, "foreign_keys": foreign_keys } for column in columns: column["primary_key"] = column['name'] in primary_keys return table_info except pyodbc.Error as e: logging.error(f"Error retrieving table info: {e}") raise - Helper to retrieve column metadata using pyodbc cursor.columns().
def _get_columns(conn, cat: str, sch: str, table:str): with conn.cursor() as cursor: ret = [] for row in cursor.columns(table=table, catalog=cat, schema=sch): ret.append({ "name":row[3], "type":row[5], "column_size": row[6], # "decimal_digits":row[8], "num_prec_radix":row[9], "nullable":False if row[10]==0 else True, "default":row[12] }) return ret - Helper to retrieve primary key constraint info using pyodbc cursor.primaryKeys().
def _get_pk_constraint(conn, cat: str, sch: str, table:str): with conn.cursor() as cursor: ret = None rs = cursor.primaryKeys(table=table, catalog=cat, schema=sch).fetchall() if len(rs) > 0: ret = { "constrained_columns": [row[3] for row in rs], "name": rs[0][5] } return ret - Helper to retrieve foreign key constraints using pyodbc cursor.foreignKeys(), grouping by FK name.
def _get_foreign_keys(conn, cat: str, sch: str, table:str): def fkey_rec(): return { "name": None, "constrained_columns": [], "referred_cat": None, "referred_schem": None, "referred_table": None, "referred_columns": [], "options": {}, } fkeys = defaultdict(fkey_rec) with conn.cursor() as cursor: rs = cursor.foreignKeys(foreignTable=table, foreignCatalog=cat, foreignSchema=sch) for row in rs: rec = fkeys[row[11]] #.FK_NAME rec["name"] = row[11] #.FK_NAME c_cols = rec["constrained_columns"] c_cols.append(row[7]) #.FKCOLUMN_NAME) r_cols = rec["referred_columns"] r_cols.append(row[3]) #.PKCOLUMN_NAME) if not rec["referred_table"]: rec["referred_table"] = row[2] #.PKTABLE_NAME rec["referred_schem"] = row[1] #.PKTABLE_OWNER rec["referred_cat"] = row[0] #.PKTABLE_CAT return list(fkeys.values()) - Utility to create pyodbc connection using env vars or provided creds.
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)