get_foreign_keys
Retrieve foreign key constraints for a specified table in CockroachDB to understand table relationships and enforce data integrity.
Instructions
Get foreign key constraints for a table.
Args:
table: Table name (schema.table or just table).
Returns:
Foreign key information.
Input Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| table | Yes |
Implementation Reference
- src/cockroachdb_mcp/server.py:205-219 (handler)MCP tool handler and registration for 'get_foreign_keys'. Decorated with @mcp.tool(), handles errors and delegates to tables.get_foreign_keys.@mcp.tool() async def get_foreign_keys(table: str) -> dict[str, Any]: """Get foreign key constraints for a table. Args: table: Table name (schema.table or just table). Returns: Foreign key information. """ try: return await tables.get_foreign_keys(table) except Exception as e: return {"status": "error", "error": str(e)}
- Core implementation of get_foreign_keys: parses table name, queries information_schema.table_constraints and related tables to fetch foreign key details, formats and returns the results.async def get_foreign_keys(table: str) -> dict[str, Any]: """Get foreign key constraints for a table. Args: table: Table name (schema.table or just table). Returns: Foreign key information. """ conn = await connection_manager.ensure_connected() # Parse schema and table name if "." in table: schema, table_name = table.rsplit(".", 1) else: schema = "public" table_name = table try: async with conn.cursor() as cur: await cur.execute( """ SELECT tc.constraint_name, kcu.column_name, ccu.table_schema AS foreign_table_schema, ccu.table_name AS foreign_table_name, ccu.column_name AS foreign_column_name FROM information_schema.table_constraints AS tc JOIN information_schema.key_column_usage AS kcu ON tc.constraint_name = kcu.constraint_name AND tc.table_schema = kcu.table_schema JOIN information_schema.constraint_column_usage AS ccu ON ccu.constraint_name = tc.constraint_name AND ccu.table_schema = tc.table_schema WHERE tc.constraint_type = 'FOREIGN KEY' AND tc.table_schema = %s AND tc.table_name = %s """, (schema, table_name), ) rows = await cur.fetchall() foreign_keys = [] for row in rows: foreign_keys.append( { "constraint_name": row.get("constraint_name"), "column": row.get("column_name"), "references_schema": row.get("foreign_table_schema"), "references_table": row.get("foreign_table_name"), "references_column": row.get("foreign_column_name"), } ) return { "schema": schema, "table": table_name, "foreign_keys": foreign_keys, "count": len(foreign_keys), } except Exception as e: return {"status": "error", "error": str(e)}