CockroachDB MCP Server

Instructions

Args:
    table: Table name (schema.table or just table).

Returns:
    Foreign key information.

Implementation Reference

• src/cockroachdb_mcp/tools/tables.py:387-450 (handler)

  Core handler function that queries information_schema for foreign key constraints on the given table and returns structured FK information.

  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)}

• src/cockroachdb_mcp/server.py:205-219 (registration)

  MCP tool registration using @mcp.tool() decorator, which delegates to the core handler in tools.tables.

  @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)}